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:

guestbook.py
from anom import Model, props
from bottle import get, post, redirect, request, run, view

Then we define a GuestbookEntry model:

guestbook.py
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:

guestbook.py
@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:

guestbook.py
@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:

guestbook.py
@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:

guestbook.py
run(host="localhost", port=8080, debug=True, reloader=True)

Here is the template we used to render the listing:

views/index.tpl
<!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.

models.py
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).

models.py
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:

models.py
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:

models.py
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:

models.py
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.

session.py
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:

session.py
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:

session.py
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:

blog.py
@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:

blog.py
@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:

blog.py
@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.

blog.py
@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:

blog.py
run(host="localhost", port=8080, debug=True, reloader=True)