2. Create database models

At this point we should create our models. In a nutshell, models represent data and its underlying storage mechanisms in an application.

We will use a relational database and SQLAlchemy’s ORM layer to access our data.

Create and edit models/user.py

Our application will consist of two tables:

  • users - stores all users for our application
  • entries - stores our blog entries

We should assume that our users might use some Unicode characters, so we need to import the Unicode datatype from SQLAlchemy. We will also need a DateTime field to timestamp our blog entries.

Let’s first create models/user.py.

$ touch pyramid_blogr/models/user.py

Add the following code to models/user.py.

1
2
3
4
5
6
7
8
9
import datetime #<- will be used to set default dates on models
from pyramid_blogr.models.meta import Base  #<- we need to import our sqlalchemy metadata from which model classes will inherit
from sqlalchemy import (
    Column,
    Integer,
    Unicode,     #<- will provide Unicode field
    UnicodeText, #<- will provide Unicode text field
    DateTime,    #<- time abstraction field
)

Make a copy of models/user.py as models/blog_record.py. We will need these imports in both modules.

$ cp pyramid_blogr/models/user.py pyramid_blogr/models/blog_record.py

The alchemy scaffold in Pyramid provides an example model class MyModel that we don’t need, as well as code that creates an index, so let’s remove the file models/mymodel.py.

$ rm pyramid_blogr/models/mymodel.py

Now our project structure should look like this.

pyramid_blogr/
......
├── models      <- model definitions aka data sources (often RDBMS or noSQL)
│     ├── __init__.py <- database engine initialization
│     ├── blog_record.py
│     ├── meta.py <- database sqlalchemy metadata object
│     └── user.py
......

Database session management

Note

To learn how to use SQLAlchemy, please consult its Object Relational Tutorial.

If you are new to SQLAlchemy or ORM’s, you are probably wondering what the code from models/__init__.py does.

To explain we need to start reading it from the includeme() part.

52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

def includeme(config):
    """
    Initialize the model for a Pyramid app.

    Activate this setup using ``config.include('pyramid_blogr.models')``.

    """
    settings = config.get_settings()
    settings['tm.manager_hook'] = 'pyramid_tm.explicit_manager'

    # use pyramid_tm to hook the transaction lifecycle to the request
    config.include('pyramid_tm')

    # use pyramid_retry to retry a request when transient exceptions occur
    config.include('pyramid_retry')

    session_factory = get_session_factory(get_engine(settings))
    config.registry['dbsession_factory'] = session_factory

    # make request.dbsession available for use in Pyramid
    config.add_request_method(
        # r.tm is the transaction manager used by pyramid_tm
        lambda r: get_tm_session(session_factory, r.tm),
        'dbsession',
        reify=True
    )

The first line defines a special function called includeme it will be picked up by pyramid on runtime and will ensure that on every request, the request object will have a dbsession propery attached that will point to SQLAlchemy’s session object.

The function also imports pyramid_tm - it is Pyramid’s transaction manager that will be attached to our request object as tm property, it will be managing our dbsession objects behavior.

We will use it to interact with the database and persist our changes to the database. It is thread-safe, meaning that it will handle multiple requests at the same time in a safe way, and our code from different views will not impact other requests. It will also open and close database connections for us transparently when needed.

What does transaction manager do?

WHOA THIS SOUNDS LIKE SCARY MAGIC!!

Note

It’s not.

OK, so while it might sound complicated, in practice it’s very simple and saves a developer a lot of headaches with managing transactions inside an application.

Here’s how the transaction manager process works:

  • A transaction is started when a browser request invokes our view code.
  • Some operations take place; for example, database rows are inserted or updated in our datastore.
    • If everything went fine, we don’t need to commit our transaction explictly; the transaction manager will do this for us.
    • If some unhandled exception occurred, we usually want to roll back all the changes and queries that were sent to our datastore; the transaction manager will handle this for us.

What are the implications of this?

Imagine you have an application that sends a confirmation email every time a user registers. A user, Nephthys, inputs the data to register, and we send Nephthys a nice welcome email and maybe an activation link, but during registration flow, something unexpected happens and the code errors out.

It is very common in this situation that the user would get a welcome email, but in reality their profile was never persisted in the database. With packages like pyramid_mailer it is perfectly possible to delay email sending until after the user’s information is successfully saved in the database.

Nice, huh?

Although this is a more advanced topic not covered in depth in this tutorial, the most simple explanation is that the transaction manager will make sure our data gets correctly saved if everything went smoothly, and if an error occurs then our datastore modifications are rolled back.

Adding model definitions

Note

This will make the application error out and prevent it from starting until we reach the last point of the current step and fix imports in other files. It’s perfectly normal, so don’t worry about immediate errors.

We will need two declarations of models that will replace the MyModel class that was created when we scaffolded our project.

After the import part in models/user.py add the following.

12
13
14
15
16
17
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(Unicode(255), unique=True, nullable=False)
    password = Column(Unicode(255), nullable=False)
    last_logged = Column(DateTime, default=datetime.datetime.utcnow)

After the import part in models/blog_record.py add the following.

12
13
14
15
16
17
18
class BlogRecord(Base):
    __tablename__ = 'entries'
    id = Column(Integer, primary_key=True)
    title = Column(Unicode(255), unique=True, nullable=False)
    body = Column(UnicodeText, default=u'')
    created = Column(DateTime, default=datetime.datetime.utcnow)
    edited = Column(DateTime, default=datetime.datetime.utcnow)

Now it’s time to update our models/__init__.py to include our models. This is especially handy because it ensures that SQLAlchemy mappers will pick up all of our model classes and functions, like create_all, and that the models will do what you expect.

Add these imports to the file (remember to also remove the MyModel import).

 6
 7
 8
 9
10
# import or define all models here to ensure they are attached to the
# Base.metadata prior to any initialization routines
from .user import User
from .blog_record import BlogRecord

Update database initialization script

It’s time to update our database initialization script to mirror the changes in our models package.

Open scripts/initialize_db.py. This is the file that actually gets executed when we run initialize_pyramid_blogr_db.

We want to replace the following bits:

def setup_models(dbsession):
    """
    Add or update models / fixtures in the database.

    """
    model = models.mymodel.MyModel(name='one', value=1)
    dbsession.add(model)

with this:

10
11
12
13
14
15
16
17
def setup_models(dbsession):
    """
    Add or update models / fixtures in the database.

    """

    model = models.user.User(name=u'admin', password=u'admin')
    dbsession.add(model)

When you initialize a fresh database, this will populate it with a single user, with both login and unencrypted password equal to admin.

Warning

This is just a tutorial example and production code should utilize passwords hashed with a strong one-way encryption function. You can use a package like passlib for this purpose. This is covered later in the tutorial.

The last step to get the application running is to change views/default.py MyModel class into out User model.

 9
10
11
12
13
14
15
16
@view_config(route_name='home', renderer='../templates/mytemplate.jinja2')
def my_view(request):
    try:
        query = request.dbsession.query(models.User)
        one = query.filter(models.User.name == 'one').first()
    except DBAPIError:
        return Response(db_err_msg, content_type='text/plain', status=500)
    return {'one': one, 'project': 'pyramid_blogr'}

Our application should start again if we try running the server.

$ $VENV/bin/pserve --reload development.ini

If you visit the URL http://0.0.0.0:6543 then you should see a “Pyramid is having a problem …” error message.

In case you have problems starting the application, you can see complete source code of the files we modifed below.

models/__init__.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from sqlalchemy import engine_from_config
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import configure_mappers
import zope.sqlalchemy

# import or define all models here to ensure they are attached to the
# Base.metadata prior to any initialization routines
from .user import User
from .blog_record import BlogRecord

# run configure_mappers after defining all of the models to ensure
# all relationships can be setup
configure_mappers()


def get_engine(settings, prefix='sqlalchemy.'):
    return engine_from_config(settings, prefix)


def get_session_factory(engine):
    factory = sessionmaker()
    factory.configure(bind=engine)
    return factory


def get_tm_session(session_factory, transaction_manager):
    """
    Get a ``sqlalchemy.orm.Session`` instance backed by a transaction.

    This function will hook the session to the transaction manager which
    will take care of committing any changes.

    - When using pyramid_tm it will automatically be committed or aborted
      depending on whether an exception is raised.

    - When using scripts you should wrap the session in a manager yourself.
      For example::

          import transaction

          engine = get_engine(settings)
          session_factory = get_session_factory(engine)
          with transaction.manager:
              dbsession = get_tm_session(session_factory, transaction.manager)

    """
    dbsession = session_factory()
    zope.sqlalchemy.register(
        dbsession, transaction_manager=transaction_manager)
    return dbsession


def includeme(config):
    """
    Initialize the model for a Pyramid app.

    Activate this setup using ``config.include('pyramid_blogr.models')``.

    """
    settings = config.get_settings()
    settings['tm.manager_hook'] = 'pyramid_tm.explicit_manager'

    # use pyramid_tm to hook the transaction lifecycle to the request
    config.include('pyramid_tm')

    # use pyramid_retry to retry a request when transient exceptions occur
    config.include('pyramid_retry')

    session_factory = get_session_factory(get_engine(settings))
    config.registry['dbsession_factory'] = session_factory

    # make request.dbsession available for use in Pyramid
    config.add_request_method(
        # r.tm is the transaction manager used by pyramid_tm
        lambda r: get_tm_session(session_factory, r.tm),
        'dbsession',
        reify=True
    )

models/user.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import datetime #<- will be used to set default dates on models
from pyramid_blogr.models.meta import Base  #<- we need to import our sqlalchemy metadata from which model classes will inherit
from sqlalchemy import (
    Column,
    Integer,
    Unicode,     #<- will provide Unicode field
    UnicodeText, #<- will provide Unicode text field
    DateTime,    #<- time abstraction field
)


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(Unicode(255), unique=True, nullable=False)
    password = Column(Unicode(255), nullable=False)
    last_logged = Column(DateTime, default=datetime.datetime.utcnow)

models/blog_record.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import datetime #<- will be used to set default dates on models
from pyramid_blogr.models.meta import Base  #<- we need to import our sqlalchemy metadata from which model classes will inherit
from sqlalchemy import (
    Column,
    Integer,
    Unicode,     #<- will provide Unicode field
    UnicodeText, #<- will provide Unicode text field
    DateTime,    #<- time abstraction field
)


class BlogRecord(Base):
    __tablename__ = 'entries'
    id = Column(Integer, primary_key=True)
    title = Column(Unicode(255), unique=True, nullable=False)
    body = Column(UnicodeText, default=u'')
    created = Column(DateTime, default=datetime.datetime.utcnow)
    edited = Column(DateTime, default=datetime.datetime.utcnow)

scripts/initialize_db.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
import argparse
import sys

from pyramid.paster import bootstrap, setup_logging
from sqlalchemy.exc import OperationalError

from .. import models


def setup_models(dbsession):
    """
    Add or update models / fixtures in the database.

    """

    model = models.user.User(name=u'admin', password=u'admin')
    dbsession.add(model)


def parse_args(argv):
    parser = argparse.ArgumentParser()
    parser.add_argument(
        'config_uri',
        help='Configuration file, e.g., development.ini',
    )
    return parser.parse_args(argv[1:])


def main(argv=sys.argv):
    args = parse_args(argv)
    setup_logging(args.config_uri)
    env = bootstrap(args.config_uri)

    try:
        with env['request'].tm:
            dbsession = env['request'].dbsession
            setup_models(dbsession)
    except OperationalError:
        print('''
Pyramid is having a problem using your SQL database.  The problem
might be caused by one of the following things:

1.  You may need to initialize your database tables with `alembic`.
    Check your README.txt for description and try to run it.

2.  Your database server may not be running.  Check that the
    database server referred to by the "sqlalchemy.url" setting in
    your "development.ini" file is running.
            ''')

__init__.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from pyramid.config import Configurator


def main(global_config, **settings):
    """ This function returns a Pyramid WSGI application.
    """
    with Configurator(settings=settings) as config:
        config.include('.models')
        config.include('pyramid_jinja2')
        config.include('.routes')
        config.scan()
    return config.make_wsgi_app()

views/default.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from pyramid.view import view_config
from pyramid.response import Response

from sqlalchemy.exc import DBAPIError

from .. import models


@view_config(route_name='home', renderer='../templates/mytemplate.jinja2')
def my_view(request):
    try:
        query = request.dbsession.query(models.User)
        one = query.filter(models.User.name == 'one').first()
    except DBAPIError:
        return Response(db_err_msg, content_type='text/plain', status=500)
    return {'one': one, 'project': 'pyramid_blogr'}


db_err_msg = """\
Pyramid is having a problem using your SQL database.  The problem
might be caused by one of the following things:

1.  You may need to initialize your database tables with `alembic`.
    Check your README.txt for descriptions and try to run it.

2.  Your database server may not be running.  Check that the
    database server referred to by the "sqlalchemy.url" setting in
    your "development.ini" file is running.

After you fix the problem, please restart the Pyramid application to
try it again.
"""

If our application starts correctly, you should run the initialize_pyramid_blogr_db command to generate database migrations.

# run this in the root of the project directory
$ $VENV/bin/alembic -c development.ini revision --autogenerate -m "init"

This will generate database migration file out of your models in pyramid_blogr/alembic/versions/ directory.

Example output:

2018-12-23 15:49:16,408 INFO  [alembic.runtime.migration:117][MainThread] Context impl SQLiteImpl.
2018-12-23 15:49:16,408 INFO  [alembic.runtime.migration:122][MainThread] Will assume non-transactional DDL.
2018-12-23 15:49:16,423 INFO  [alembic.autogenerate.compare:115][MainThread] Detected added table 'entries'
2018-12-23 15:49:16,423 INFO  [alembic.autogenerate.compare:115][MainThread] Detected added table 'users'
  Generating /home/ergo/workspace/pyramid_blogr/pyramid_blogr/alembic/versions/20181223_5899f27f265f.py ... done

Generated migration file might look like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
"""init

Revision ID: 5899f27f265f
Revises: 
Create Date: 2018-12-23 16:39:13.677058

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '5899f27f265f'
down_revision = None
branch_labels = None
depends_on = None

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('entries',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('title', sa.Unicode(length=255), nullable=False),
    sa.Column('body', sa.UnicodeText(), nullable=True),
    sa.Column('created', sa.DateTime(), nullable=True),
    sa.Column('edited', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_entries')),
    sa.UniqueConstraint('title', name=op.f('uq_entries_title'))
    )
    op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.Unicode(length=255), nullable=False),
    sa.Column('password', sa.Unicode(length=255), nullable=False),
    sa.Column('last_logged', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_users')),
    sa.UniqueConstraint('name', name=op.f('uq_users_name'))
    )
    # ### end Alembic commands ###

def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('users')
    op.drop_table('entries')
    # ### end Alembic commands ###

Now you can run the migration against your database.

$ $VENV/bin/alembic -c development.ini upgrade head

Example output:

2018-12-23 15:51:49,238 INFO  [alembic.runtime.migration:117][MainThread] Context impl SQLiteImpl.
2018-12-23 15:51:49,238 INFO  [alembic.runtime.migration:122][MainThread] Will assume non-transactional DDL.
2018-12-23 15:51:49,239 INFO  [alembic.runtime.migration:327][MainThread] Running upgrade  -> 4325dedd2673, init

Since your database has all the necessary user and blog tables you can populate it with admin user.

$ $VENV/bin/initialize_pyramid_blogr_db development.ini

If you start the application you should be able to see index page.

Next 3. Application routes.