Skip to main content
Donate to the Python Software Foundation or Purchase a PyCharm License to Benefit the PSF! Donate Now

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

Project description

alchemist-stack

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

Last Updated: April 23rd, 2018

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

Overview

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

Usage

Example ORM Table:

# table_example.py

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:

# model_example.py

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 = datetime.now(timezone.utc).astimezone(), 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(...)`
                `x(...)`
            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,
                                                                                timestamp=self.__timestamp,
                                                                                hex_id=hex(id(self)))

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

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

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

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

Example Repository:

# repository_example.py

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:
        """
        return '<class ExampleRepository->RepositoryBase(context={context}) at {hex_id}>'\
            .format(context=str(self.context),
                    hex_id=hex(id(self.context)))

    def create_example(self, obj: Example):
        self._create_object(obj=obj.to_orm())

    def get_example_by_id(self, example_id: int) -> Example:
        self._create_session()
        __query = self._read_object(cls=ExampleTable)
        __t = __query.with_session(self.session).get(example_id)
        self._close_session()
        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:
        self._create_session()
        __query = self._update_object(cls=ExampleTable, values=values)
        rowcount = __query.with_session(self.session)\
            .filter(ExampleTable.primary_key == example_id)\
            .update(values=values,
                    synchronize_session=synchronize_session)
        self._commit_session()
        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.

Filename, size & hash SHA256 hash help File type Python version Upload date
alchemist_stack-0.1.0.dev1-py3-none-any.whl (13.4 kB) Copy SHA256 hash SHA256 Wheel py3
alchemist_stack-0.1.0.dev1.tar.gz (11.2 kB) Copy SHA256 hash SHA256 Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page