Basic Layout

The starter files generated from choosing the sqlalchemy backend option in the cookiecutter are very basic, but they provide a good orientation for the high-level patterns common to most URL dispatch-based Pyramid projects.

Application configuration with __init__.py

A directory on disk can be turned into a Python package by containing an __init__.py file. Even if empty, this marks a directory as a Python package. We use __init__.py both as a marker, indicating the directory in which it's contained is a package, and to contain application configuration code.

Open tutorial/__init__.py. It should already contain the following:

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('pyramid_jinja2')
        config.include('.routes')
        config.include('.models')
        config.scan()
    return config.make_wsgi_app()

Let's go over this piece-by-piece. First we need some imports to support later code:

from pyramid.config import Configurator

__init__.py defines a function named main. Here is the entirety of the main function we've defined in our __init__.py:

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

When you invoke the pserve development.ini command, the main function above is executed. It accepts some settings and returns a WSGI application. (See Startup for more about pserve.)

Next in main, construct a Configurator object using a context manager:

    with Configurator(settings=settings) as config:

settings is passed to the Configurator as a keyword argument with the dictionary values passed as the **settings argument. This will be a dictionary of settings parsed from the .ini file, which contains deployment-related values, such as pyramid.reload_templates, sqlalchemy.url, and so on.

Next include Jinja2 templating bindings so that we can use renderers with the .jinja2 extension within our project.

        config.include('pyramid_jinja2')

Next include the routes module using a dotted Python path. This module will be explained in the next section.

        config.include('.routes')

Next include the package models using a dotted Python path. The exact setup of the models will be covered later.

        config.include('.models')

Note

Pyramid's pyramid.config.Configurator.include() method is the primary mechanism for extending the configurator and breaking your code into feature-focused modules.

main next calls the scan method of the configurator (pyramid.config.Configurator.scan()), which will recursively scan our tutorial package, looking for @view_config and other special decorators. When it finds a @view_config decorator, a view configuration will be registered, allowing one of our application URLs to be mapped to some code.

        config.scan()

Finally main is finished configuring things, so it uses the pyramid.config.Configurator.make_wsgi_app() method to return a WSGI application:

    return config.make_wsgi_app()

Route declarations

Open the tutorial/routes.py file. It should already contain the following:

def includeme(config):
    config.add_static_view('static', 'static', cache_max_age=3600)
    config.add_route('home', '/')

On line 2, we call pyramid.config.Configurator.add_static_view() with three arguments: static (the name), static (the path), and cache_max_age (a keyword argument).

This registers a static resource view which will match any URL that starts with the prefix /static (by virtue of the first argument to add_static_view). This will serve up static resources for us from within the static directory of our tutorial package, in this case via http://localhost:6543/static/ and below (by virtue of the second argument to add_static_view). With this declaration, we're saying that any URL that starts with /static should go to the static view; any remainder of its path (e.g., the /foo in /static/foo) will be used to compose a path to a static file resource, such as a CSS file.

On line 3, the module registers a route configuration via the pyramid.config.Configurator.add_route() method that will be used when the URL is /. Since this route has a pattern equaling /, it is the route that will be matched when the URL / is visited, e.g., http://localhost:6543/.

View declarations via the views package

The main function of a web framework is mapping each URL pattern to code (a view callable) that is executed when the requested URL matches the corresponding route. Our application uses the pyramid.view.view_config() decorator to perform this mapping.

Open tutorial/views/default.py in the views package. It should already contain the following:

from pyramid.view import view_config
from pyramid.response import Response
from sqlalchemy.exc import SQLAlchemyError

from .. import models


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


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.
"""

The important part here is that the @view_config decorator associates the function it decorates (my_view) with a view configuration, consisting of:

• a route_name (home)

• a renderer, which is a template from the templates subdirectory of the package.

When the pattern associated with the home view is matched during a request, my_view() will be executed. my_view() returns a dictionary; the renderer will use the templates/mytemplate.jinja2 template to create a response based on the values in the dictionary.

Note that my_view() accepts a single argument named request. This is the standard call signature for a Pyramid view callable.

Remember in our __init__.py when we executed the pyramid.config.Configurator.scan() method config.scan()? The purpose of calling the scan method was to find and process this @view_config decorator in order to create a view configuration within our application. Without being processed by scan, the decorator effectively does nothing. @view_config is inert without being detected via a scan.

The sample my_view() created by the cookiecutter uses a try: and except: clause to detect if there is a problem accessing the project database and provide an alternate error response. That response will include the text shown at the end of the file, which will be displayed in the browser to inform the user about possible actions to take to solve the problem.

Open tutorial/views/notfound.py in the views package to look at the second view.

from pyramid.view import notfound_view_config


@notfound_view_config(renderer='tutorial:templates/404.jinja2')
def notfound_view(request):
    request.response.status = 404
    return {}

Without repeating ourselves, we will point out the differences between this view and the previous.

1. Line 4. The notfound_view function is decorated with @notfound_view_config. This decorator registers a Not Found View using pyramid.config.Configurator.add_notfound_view().

   The renderer argument names an asset specification of tutorial:templates/404.jinja2.

2. Lines 5-7. A view callable named notfound_view is defined, which is decorated in the step above. It sets the HTTP response status code to 404. The function returns an empty dictionary to the template 404.jinja2, which accepts no parameters anyway.

Content models with the models package

In a SQLAlchemy-based application, a model object is an object composed by querying the SQL database. The models package is where the alchemy cookiecutter put the classes that implement our models.

First, open tutorial/models/meta.py, which should already contain the following:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import MetaData

# Recommended naming convention used by Alembic, as various different database
# providers will autogenerate vastly different names making migrations more
# difficult. See: https://alembic.sqlalchemy.org/en/latest/naming.html
NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s"
}

metadata = MetaData(naming_convention=NAMING_CONVENTION)
Base = declarative_base(metadata=metadata)

meta.py contains imports and support code for defining the models. We create a dictionary NAMING_CONVENTION as well for consistent naming of support objects like indices and constraints.

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import MetaData

# Recommended naming convention used by Alembic, as various different database
# providers will autogenerate vastly different names making migrations more
# difficult. See: https://alembic.sqlalchemy.org/en/latest/naming.html
NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s"
}

Next we create a metadata object from the class sqlalchemy.schema.MetaData, using NAMING_CONVENTION as the value for the naming_convention argument.

A MetaData object represents the table and other schema definitions for a single database. We also need to create a declarative Base object to use as a base class for our models. Our models will inherit from this Base, which will attach the tables to the metadata we created, and define our application's database schema.

metadata = MetaData(naming_convention=NAMING_CONVENTION)
Base = declarative_base(metadata=metadata)

Next open tutorial/models/mymodel.py, which should already contain the following:

from sqlalchemy import (
    Column,
    Index,
    Integer,
    Text,
)

from .meta import Base


class MyModel(Base):
    __tablename__ = 'models'
    id = Column(Integer, primary_key=True)
    name = Column(Text)
    value = Column(Integer)


Index('my_index', MyModel.name, unique=True, mysql_length=255)

Notice we've defined the models as a package to make it straightforward for defining models in separate modules. To give a simple example of a model class, we have defined one named MyModel in mymodel.py:

class MyModel(Base):
    __tablename__ = 'models'
    id = Column(Integer, primary_key=True)
    name = Column(Text)
    value = Column(Integer)

Our example model does not require an __init__ method because SQLAlchemy supplies for us a default constructor, if one is not already present, which accepts keyword arguments of the same name as that of the mapped attributes.

Note

Example usage of MyModel:

johnny = MyModel(name="John Doe", value=10)

The MyModel class has a __tablename__ attribute. This informs SQLAlchemy which table to use to store the data representing instances of this class.

Finally, open tutorial/models/__init__.py, which should already contain the following:

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 .mymodel import MyModel  # flake8: noqa

# 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, request=None):
    """
    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:

      .. code-block:: python

          import transaction

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

    This function may be invoked with a ``request`` kwarg, such as when invoked
    by the reified ``.dbsession`` Pyramid request attribute which is configured
    via the ``includeme`` function below. The default value, for backwards
    compatibility, is ``None``.

    The ``request`` kwarg is used to populate the ``sqlalchemy.orm.Session``'s
    "info" dict.  The "info" dict is the official namespace for developers to
    stash session-specific information.  For more information, please see the
    SQLAlchemy docs:
    https://docs.sqlalchemy.org/en/stable/orm/session_api.html#sqlalchemy.orm.session.Session.params.info

    By placing the active ``request`` in the "info" dict, developers will be
    able to access the active Pyramid request from an instance of an SQLAlchemy
    object in one of two ways:

    - Classic SQLAlchemy. This uses the ``Session``'s utility class method:

      .. code-block:: python

          from sqlalchemy.orm.session import Session as sa_Session

          dbsession = sa_Session.object_session(dbObject)
          request = dbsession.info["request"]

    - Modern SQLAlchemy. This uses the "Runtime Inspection API":

      .. code-block:: python

          from sqlalchemy import inspect as sa_inspect

          dbsession = sa_inspect(dbObject).session
          request = dbsession.info["request"]
    """
    dbsession = session_factory(info={"request": request})
    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('tutorial.models')``.

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

    # Use ``pyramid_tm`` to hook the transaction lifecycle to the request.
    # Note: the packages ``pyramid_tm`` and ``transaction`` work together to
    # automatically close the active database session after every request.
    # If your project migrates away from ``pyramid_tm``, you may need to use a
    # Pyramid callback function to close the database session after each
    # request.
    config.include('pyramid_tm')

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

    # hook to share the dbengine fixture in testing
    dbengine = settings.get('dbengine')
    if not dbengine:
        dbengine = get_engine(settings)

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

    # make request.dbsession available for use in Pyramid
    def dbsession(request):
        # hook to share the dbsession fixture in testing
        dbsession = request.environ.get('app.dbsession')
        if dbsession is None:
            # request.tm is the transaction manager used by pyramid_tm
            dbsession = get_tm_session(
                session_factory, request.tm, request=request
            )
        return dbsession

    config.add_request_method(dbsession, reify=True)

Our models/__init__.py module defines the primary API we will use for configuring the database connections within our application, and it contains several functions we will cover below.

As we mentioned above, the purpose of the models.meta.metadata object is to describe the schema of the database. This is done by defining models that inherit from the Base object attached to that metadata object. In Python, code is only executed if it is imported, and so to attach the models table defined in mymodel.py to the metadata, we must import it. If we skip this step, then later, when we run sqlalchemy.schema.MetaData.create_all(), the table will not be created because the metadata object does not know about it!

Another important reason to import all of the models is that, when defining relationships between models, they must all exist in order for SQLAlchemy to find and build those internal mappings. This is why, after importing all the models, we explicitly execute the function sqlalchemy.orm.configure_mappers(), once we are sure all the models have been defined and before we start creating connections.

Next we define several functions for connecting to our database. The first and lowest level is the get_engine function. This creates an SQLAlchemy database engine using sqlalchemy.engine_from_config() from the sqlalchemy.-prefixed settings in the development.ini file's [app:main] section. This setting is a URI (something like sqlite://).

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

The function get_session_factory accepts an SQLAlchemy database engine, and creates a session_factory from the SQLAlchemy class sqlalchemy.orm.session.sessionmaker. This session_factory is then used for creating sessions bound to the database engine.

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

The function get_tm_session registers a database session with a transaction manager, and returns a dbsession object. With the transaction manager, our application will automatically issue a transaction commit after every request, unless an exception is raised, in which case the transaction will be aborted.

def get_tm_session(session_factory, transaction_manager, request=None):
    """
    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:

      .. code-block:: python

          import transaction

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

    This function may be invoked with a ``request`` kwarg, such as when invoked
    by the reified ``.dbsession`` Pyramid request attribute which is configured
    via the ``includeme`` function below. The default value, for backwards
    compatibility, is ``None``.

    The ``request`` kwarg is used to populate the ``sqlalchemy.orm.Session``'s
    "info" dict.  The "info" dict is the official namespace for developers to
    stash session-specific information.  For more information, please see the
    SQLAlchemy docs:
    https://docs.sqlalchemy.org/en/stable/orm/session_api.html#sqlalchemy.orm.session.Session.params.info

    By placing the active ``request`` in the "info" dict, developers will be
    able to access the active Pyramid request from an instance of an SQLAlchemy
    object in one of two ways:

    - Classic SQLAlchemy. This uses the ``Session``'s utility class method:

      .. code-block:: python

          from sqlalchemy.orm.session import Session as sa_Session

          dbsession = sa_Session.object_session(dbObject)
          request = dbsession.info["request"]

    - Modern SQLAlchemy. This uses the "Runtime Inspection API":

      .. code-block:: python

          from sqlalchemy import inspect as sa_inspect

          dbsession = sa_inspect(dbObject).session
          request = dbsession.info["request"]
    """
    dbsession = session_factory(info={"request": request})
    zope.sqlalchemy.register(
        dbsession, transaction_manager=transaction_manager
    )
    return dbsession

Finally, we define an includeme function, which is a hook for use with pyramid.config.Configurator.include() to activate code in a Pyramid application add-on. It is the code that is executed above when we ran config.include('.models') in our application's main function. This function will take the settings from the application, create an engine, and define a request.dbsession property, which we can use to do work on behalf of an incoming request to our application.

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

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

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

    # Use ``pyramid_tm`` to hook the transaction lifecycle to the request.
    # Note: the packages ``pyramid_tm`` and ``transaction`` work together to
    # automatically close the active database session after every request.
    # If your project migrates away from ``pyramid_tm``, you may need to use a
    # Pyramid callback function to close the database session after each
    # request.
    config.include('pyramid_tm')

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

    # hook to share the dbengine fixture in testing
    dbengine = settings.get('dbengine')
    if not dbengine:
        dbengine = get_engine(settings)

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

    # make request.dbsession available for use in Pyramid
    def dbsession(request):
        # hook to share the dbsession fixture in testing
        dbsession = request.environ.get('app.dbsession')
        if dbsession is None:
            # request.tm is the transaction manager used by pyramid_tm
            dbsession = get_tm_session(
                session_factory, request.tm, request=request
            )
        return dbsession

    config.add_request_method(dbsession, reify=True)

That's about all there is to it regarding models, views, and initialization code in our stock application.

The Index import and the Index object creation in mymodel.py is not required for this tutorial, and will be removed in the next step.

Tests

The project contains a basic structure for a test suite using pytest. The structure is covered later in Adding Tests.