Examples¶
Guestbook (Bottle)¶
This example uses anom and Bottle to build a simple guestbook application.
Running¶
To run this example, clone the anom repository to your local machine:
git clone https://github.com/Bogdanp/anom-py
Then cd
into the examples/guestbook
folder and follow the
instructions in the README.md
file. To read the un-annotated
source code online click here.
Annotated Source Code¶
First we import Model
and props
from anom
and various bottle
functions:
from anom import Model, props
from bottle import get, post, redirect, request, run, view
Then we define a GuestbookEntry
model:
class GuestbookEntry(Model):
author = props.String(optional=True)
message = props.Text()
created_at = props.DateTime(indexed=True, auto_now_add=True)
created_at
has its indexed
option set to True
so that we
can sort guestbook entries in descending order when we list them.
Since the other properties are never used for filtering or sorting,
they are left unindexed.
Next up, we define our index route:
@get("/")
@view("index")
def index():
cursor = request.params.cursor
pages = GuestbookEntry.query().order_by(-GuestbookEntry.created_at).paginate(page_size=1, cursor=cursor)
return {"pages": pages}
We paginate over the guestbook entry items one item per page for
convenience when testing and we make it possible to pass in a
cursor
query parameter. Bottle returns None
when a query
parameter does not exist which anom interprets as a cursor for the
first page of results.
Then we define a route to create new entries:
@post("/sign")
def sign():
author = request.forms.author
message = request.forms.message
if not author or not message:
return "<em>You must provide a message!</em>"
GuestbookEntry(author=author, message=message).put()
return redirect("/")
And a route to delete existing entries:
@post("/delete/<entry_id:int>")
def delete(entry_id):
entry = GuestbookEntry.get(entry_id)
if not entry:
return "<h1>Entry not found.</h1>"
entry.delete()
return redirect("/")
Finally, we run the server:
run(host="localhost", port=8080, debug=True, reloader=True)
Here is the template we used to render the listing:
<!doctype html>
<html lang="en-us">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
<title>Guestbook Example</title>
<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/4.0.0-alpha.6/css/bootstrap.min.css" integrity="sha384-rwoIResjU2yc3z8GV/NPeZWAv56rSmLldC3R/AZzGRnGxQQKnKkoFVhFQhNUwEyJ" crossorigin="anonymous">
</head>
<body>
<div class="container">
<h1>Guestbook</h1>
<br/>
<div class="row">
<div class="col-4">
<h4>Sign guestbook</h4>
<br/>
<form action="/sign" method="POST">
<div class="form-group">
<label for="author">Name:</label>
<input id="author" name="author" class="form-control" />
</div>
<div class="form-group">
<label for="message">Message:</label>
<textarea id="message" name="message" rows="8" class="form-control"></textarea>
</div>
<button type="submit" class="btn btn-primary">Sign guestbook</button>
</form>
</div>
<div class="col-8">
<h4>Entries:</h4>
<br/>
% for entry in pages.fetch_next_page():
<div class="row">
<div class="col-12">
<p>{{entry.message}}</p>
<hr>
<form action="/delete/{{entry.key.int_id}}" method="POST">
<h6>
Signed by <strong>{{entry.author or 'anonymous'}}</strong> on <em>{{entry.created_at}}</em>.
<button type="submit" onclick="return confirm('Are you sure?');" class="btn btn-danger btn-sm">
Delete
</button>
</h6>
</form>
</div>
</div>
% end
% if pages.has_more:
<div class="row">
<div class="col-12">
<hr>
<a href="/?cursor={{pages.cursor}}" class="btn btn-secondary btn-sm">Next page</a>
</div>
</div>
% end
</div>
</div>
</div>
<script src="https://code.jquery.com/jquery-3.1.1.slim.min.js" integrity="sha384-A7FZj7v+d/sdmMqp/nOQwliLvUsJfDHW+k9Omg/a/EheAdgtzNs3hpfag6Ed950n" crossorigin="anonymous"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/tether/1.4.0/js/tether.min.js" integrity="sha384-DztdAPBWPRXSA/3eYEEUWrWCy7G5KFbe8fFjk5JAIxUYHKkDx6Qin1DkWx51bBrb" crossorigin="anonymous"></script>
<script src="https://maxcdn.bootstrapcdn.com/bootstrap/4.0.0-alpha.6/js/bootstrap.min.js" integrity="sha384-vBWWzlZJ8ea9aCX4pEW3rVHjgjt7zpkNpZk+02D9phzyeVkE+jo0ieGizqPLForn" crossorigin="anonymous"></script>
</body>
</html>
Blog (Bottle)¶
This example uses anom and Bottle to build a simple blog.
Running¶
To run this example, clone the anom repository to your local machine:
git clone https://github.com/Bogdanp/anom-py
Then cd
into the examples/blog
folder and follow the
instructions in the README.md
file. To read the un-annotated
source code online click here.
Annotated Source Code¶
Models¶
In order to make password hashing transparent, we define a custom
Password
property that hashes values automatically every time
they are assigned to an entity.
ctx = CryptContext(schemes=["sha256_crypt"])
class Password(props.String):
def validate(self, value):
return ctx.hash(super().validate(value))
Then we declare a polymorphic user model. Polymorphic models allow us
to define class hierarchies that are all stored under the same
Datastore kind, this means that User
entities and entities of its
subclasses will each be stored under the User
kind in Datastore
(as opposed to standard inheritance, where each subclass would get its
own kind).
class User(Model, poly=True):
username = props.String(indexed=True)
password = Password()
created_at = props.DateTime(indexed=True, auto_now_add=True)
updated_at = props.DateTime(indexed=True, auto_now=True)
@classmethod
def login(cls, username, password):
user = User.query().where(User.username == username).get()
if user is None:
return None
if not ctx.verify(password, user.password):
return None
if ctx.needs_update(user.password):
user.password = password
return user.put()
return user
@property
def permissions(self):
return ()
Next, we define Editor
and Reader
users:
class Editor(User):
@property
def permissions(self):
return ("create", "read", "edit", "delete")
class Reader(User):
@property
def permissions(self):
return ("read",)
And a model for Post
entities:
class Post(Model):
author = props.Key(indexed=True, kind=User)
title = props.String()
def __compute_slug(self):
if self.title is None:
return None
return slugify(self.title)
def __compute_body(self):
if self.body is None:
return None
return markdown(self.body)
slug = props.Computed(__compute_slug)
body = props.Text()
body_markdown = props.Computed(__compute_body)
tags = props.String(indexed=True, repeated=True)
created_at = props.DateTime(indexed=True, auto_now_add=True)
updated_at = props.DateTime(indexed=True, auto_now=True)
The repreated tags
property on posts is there so that we can
filter post queries by one or many tags at a time.
Finally, we define and call a method to populate the database with an editor and a reader user on first run:
def init_database():
users = list(User.query().run(keys_only=True))
if users:
return
Editor(username="editor", password="editor").put()
Reader(username="viewer", password="viewer").put()
init_database()
Sessions¶
We define a Session
model that is used to assign random session
ids to individual users when they login.
class Session(Model):
user = props.Key(kind=User)
@classmethod
def create(cls, user):
session = cls(user=user)
session.key = Key(Session, str(uuid4()))
return session.put()
Here we use the uuid4
function to generate random session ids and
assign them as custom ids to new Session
entities. This way we
can store the session id on the user’s computer using a cookie and
then read that cookie back to look up the session via a get
call rather than a Query
in the user_required
decorator:
def user_required(*permissions):
def decorator(fn):
def wrapper(*args, **kwargs):
session_id = request.get_cookie("sid")
if not session_id:
return redirect("/login")
session = Session.get(session_id)
if not session:
return redirect("/login")
user = session.user.get()
if user is None:
return redirect("/login")
for permission in permissions:
if permission not in user.permissions:
return abort(403)
return fn(user, *args, **kwargs)
return wrapper
return decorator
Finally we define a helper for creating Session
objects and
setting the session id cookie:
def create_session(user):
session = Session.create(user)
response.set_cookie("sid", session.key.str_id)
Routes¶
We define a route to list blog posts, optionally allowing them to be filtered by tag:
@get("/")
@view("index")
@user_required()
def index(user):
cursor, tag = request.params.cursor, request.params.tag
posts = Post.query().order_by(-Post.created_at)
if tag:
posts = posts.where(Post.tags == tag)
pages = posts.paginate(page_size=2, cursor=cursor)
return {"user": user, "pages": pages}
And a route that lets users view individual posts by slug:
@get("/posts/<slug>")
@view("post")
@user_required("read")
def view_post(user, slug):
post = Post.query().where(Post.slug == slug).get()
if post is None:
return abort(404)
return {"post": post, "embedded": False}
One that lets users with create
permissions to create posts:
@get("/new")
@view("create")
@user_required("create")
def create(user):
return {}
@post("/new")
@view("create")
@user_required("create")
def do_create(user):
title = request.forms.title
tags = [tag.strip() for tag in request.forms.tags.split(",") if tag.strip()]
body = request.forms.body
post = Post(author=user, title=title, tags=tags, body=body).put()
return redirect("/posts/" + post.slug)
And one that logs users in. This is the route anonymous users are
taken to when they visit routes that have been decorated with
user_required
.
@get("/login")
@view("login")
def login():
return {}
@post("/login")
@view("login")
def do_login():
username = request.forms.username
password = request.forms.password
user = User.login(username, password)
if not user:
return {"error": "Invalid credentials."}
create_session(user)
return redirect("/")
Finally, we run the server:
run(host="localhost", port=8080, debug=True, reloader=True)