ordr3/ordr3/adapters.py

""" sqlalchemy metadata configuration """

from datetime import datetime

import zope.sqlalchemy
from sqlalchemy import (
    Enum,
    Text,
    Float,
    Table,
    Column,
    Integer,
    DateTime,
    ForeignKey,
    engine_from_config,
)
from sqlalchemy.orm import mapper, relationship, sessionmaker
from sqlalchemy.schema import MetaData

from . import repo, models

# Recommended naming convention used by Alembic, as various different database
# providers will autogenerate vastly different names making migrations more
# difficult. See: http://alembic.zzzcomputing.com/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)

order_table = Table(
    "orders",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("cas_description", Text, nullable=False),
    Column("catalog_nr", Text, nullable=False),
    Column("vendor", Text, nullable=False),
    Column("category", Enum(models.OrderCategory), nullable=False),
    Column("package_size", Text, nullable=False),
    Column("unit_price", Float, nullable=False),
    Column("currency", Text, nullable=False, default="EUR"),
    Column("amount", Integer, nullable=False),
    Column("account", Text, nullable=False, default=""),
    Column("comment", Text, nullable=False, default=""),
    # redundant properties, could be determined from orders log
    Column("created_on", DateTime, nullable=False, default=datetime.utcnow),
    Column("created_by", Text, nullable=True, index=True),
    Column("status", Enum(models.OrderStatus), nullable=True),
)


log_table = Table(
    "logs",
    metadata,
    Column("order_id", Integer, ForeignKey("orders.id"), primary_key=True),
    Column("status", Enum(models.OrderStatus), primary_key=True),
    Column("date", DateTime, primary_key=True, default=datetime.utcnow),
    Column("by", Text, nullable=False),
)

vendor_table = Table(
    "vendors",
    metadata,
    Column("term", Text, primary_key=True),
    Column("name", Text, index=True),
)

user_table = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("username", Text, nullable=False, index=True),
    Column("first_name", Text, nullable=False),
    Column("last_name", Text, nullable=False),
    Column("email", Text, nullable=False, index=True),
    Column("password", Text, nullable=False),
    Column("role", Enum(models.UserRole)),
)

reset_token_table = Table(
    "reset_tokens",
    metadata,
    Column("token", Text, primary_key=True),
    Column("user_id", Integer, nullable=False),
    Column("valid_until", DateTime, nullable=False),
)


def start_mappers():
    """ maps data base tables to model objects """
    mapper(
        models.OrderItem,
        order_table,
        properties={
            "log": relationship(
                models.LogEntry, backref="order", order_by=log_table.c.date
            )
        },
    )
    mapper(models.LogEntry, log_table)
    mapper(models.Vendor, vendor_table)
    mapper(models.User, user_table)
    mapper(models.PasswordResetToken, reset_token_table)


def get_engine(settings, prefix="sqlalchemy."):
    """ returns a sqlalchemy engine from a settings dict """
    return engine_from_config(settings, prefix)


def get_session_factory(engine):
    """ returns a sqlalchemy session factory for a db enging """
    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 get_repo_with_session(session_factory, request):
    """ returns an sql alchemy repository with database session configured """
    # request.tm is the transaction manager used by pyramid_tm
    session = get_tm_session(session_factory, request.tm)
    return repo.SqlAlchemyRepository(session)


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

    Activate this setup using ``config.include('ordr3.adapters')``.

    """
    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(
        lambda r: get_repo_with_session(session_factory, r), "repo", reify=True
    )

    start_mappers()