Skip to main content

A Thread-Safe, Multi-Session/Multi-Connection Model-Repository-Context base for SQL Alchemy

Project description


Package Author: H.D. 'Chip' McCullough IV

Last Updated: April 23rd, 2018

A Flexible Model-Repository-Database stack for use with SQL Alchemy


Alchemist Stack is intended to be a thread-safe, multi-session/multi-connection


Example ORM Table:


from alchemist_stack.repository.models import Base
from sqlalchemy import Column, Integer, DateTime

class ExampleTable(Base):
    __tablename__ = 'example'

    primary_key = Column('id', Integer, primary_key=True)
    timestamp = Column(DateTime(timezone=True), nullable=False)

    def __repr__(self):
        return '<Example(timestamp={timestamp})>'.format(timestamp=self.timestamp)

Example Model:


from tables.table_example import ExampleTable

from datetime import datetime, timezone
from typing import TypeVar

E = TypeVar('E', bound="Example")

class Example(object):
        Example Model class.

    def __init__(self, timestamp: datetime =, primary_key: int = None,
                 *args, **kwargs):
        self.__pk = primary_key
        self.__timestamp = timestamp
        self.__args = args
        self.__kwargs = kwargs

    def __call__(self, *args, **kwargs) -> ExampleTable:
            Called when an instance of Example is called, e.g.:
                `x = Example(...)`
            This is equivalent to calling `to_orm()` on the object instance.

        :returns: The ORM of the Example.
        :rtype: ExampleTable
        return self.to_orm()

    def __repr__(self) -> str:
            A detailed String representation of Example.

        :returns: String representation of Example object.
        return '<class Test(pk={pk} timestamp={timestamp}) at {hex_id}>'.format(pk=self.__pk,

    def id(self) -> int:
        return self.__pk

    def timestamp(self) -> datetime:
        return self.__timestamp

    def to_orm(self) -> ExampleTable:
        return ExampleTable(primary_key=self.__pk, timestamp=self.__timestamp)

    def from_orm(cls, obj: ExampleTable) -> E:
        return cls(timestamp=obj.timestamp, primary_key=obj.primary_key)

Example Repository:


from alchemist_stack.context import Context
from alchemist_stack.repository import RepositoryBase
from models.model_example import Example
from tables.table_example import ExampleTable

class ExampleRepository(RepositoryBase):


    def __init__(self, context: Context, *args, **kwargs):
            Test Repository Constructor
        :param database: The Database object containing the engine used to connect to the database.
        :param args: Additional Arguments
        :param kwargs: Additional Keyword Arguments
        super().__init__(context=context, *args, **kwargs)

    def __repr__(self):

        return '<class ExampleRepository->RepositoryBase(context={context}) at {hex_id}>'\

    def create_example(self, obj: Example):

    def get_example_by_id(self, example_id: int) -> Example:
        __query = self._read_object(cls=ExampleTable)
        __t = __query.with_session(self.session).get(example_id)
        if isinstance(__t, ExampleTable):
            return Example.from_orm(__t)

    def update_example_by_id(self, example_id: int, values: dict, synchronize_session: str = 'evaluate') -> int:
        __query = self._update_object(cls=ExampleTable, values=values)
        rowcount = __query.with_session(self.session)\
            .filter(ExampleTable.primary_key == example_id)\
        return rowcount

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for alchemist-stack, version 0.1.0.dev1
Filename, size File type Python version Upload date Hashes
Filename, size alchemist_stack-0.1.0.dev1-py3-none-any.whl (13.4 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size alchemist_stack-0.1.0.dev1.tar.gz (11.2 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring DigiCert DigiCert EV certificate Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page