Advanced query manager for SQLAlchemy
Project description
Managers for SQLAlchemy.
Manager for model, methods were added during runtime to query.
Installation
$ [sudo] pip install advanced-sqlalchemy-manager
Documentation
ManagedQuery
Managed query that replaces sqlalchemy.query class
Example:
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String
from alchmanager import ManagedQuery, BaseQueryManager
engine = create_engine('sqlite:///:memory:')
session_factory = sessionmaker(query_cls=ManagedQuery, bind=engine)
session = session_factory()
Base = declarative_base()
class Person(Base):
__tablename__ = 'persons'
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
age = Column(Integer, nullable=False)
class PersonQueryManager(BaseQueryManager):
__model__ = Person
@staticmethod
def older_than(query: ManagedQuery, age: int) -> ManagedQuery:
return query.filter(Person.age > age)
@staticmethod
def younger_than(query: ManagedQuery, age: int):
return query.filter(Person.age < age)
@staticmethod
def first_of_exact_age(query: ManagedQuery, age: int):
return query.filter(Person.age == age).first()
filtered_persons = session.query(Person).older_than(30).filter(Person.name.contains('_')).younger_than(60).all()
person_25_years_old = session.query(Person).first_of_exact_age(25)
ManagedSession
Managed session.
Use decorator load_manager()
to register query managers into session.
Registered that way session managers will be usable on any model.
Example:
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, Boolean
from alchmanager import ManagedQuery, ManagedSession, BaseQueryManager
engine = create_engine('sqlite:///:memory:')
session_factory = sessionmaker(class_=ManagedSession, bind=engine)
session = session_factory()
Base = declarative_base()
class Person(Base):
__tablename__ = 'persons'
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
age = Column(Integer, nullable=False)
class Book(Base):
__tablename__ = 'books'
id = Column(Integer, primary_key=True)
title = Column(String)
is_public = Column(Boolean, nullable=False, default=False)
@session.load_manager()
class BookQueryManager(BaseQueryManager):
@staticmethod
def is_book_public(query: ManagedQuery) -> ManagedQuery:
return query.filter(Book.is_public.is_(True))
count_of_filtered_books = session.query(Book).is_book_public().count()
# This will produce broken query because is_public does not exists in Person model
persons = session.query(Person).is_book_public().count()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for advanced-sqlalchemy-manager-0.1.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 2b6bf7e1d76b428eb8822fa42fd2c24e2c984b0d1c8139ba2db2ecc7d127d179 |
|
MD5 | 5a2e40d8757ce767ecb00d6ba08c33a3 |
|
BLAKE2b-256 | 3513e96d0cd09e1d97767bcb59ec40c0e313dce40761f755f0efafaaa8011fe9 |
Close
Hashes for advanced_sqlalchemy_manager-0.1.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 80b35404d6cbe42dcf771f9ed316238fdaeac0ca70842c5d821e0182e638d29a |
|
MD5 | 504c42b6022841b757b48940b29334f6 |
|
BLAKE2b-256 | d43cdfa50b144399d9fe679432dbf88d4a0325ef0c5be969bc182e5a0930733a |