Wrapper around SQLAlchemy AsyncSession with built-in query execution profiling and connection context management.

---

Installation

pip install gadsqlalchemy

Usage

# logger: sqlalchemy.profiler

from gadsqlalchemy import Sqlalchemy, Base, CRUD

alchemy = Sqlalchemy("postgresql+asyncpg://postgres:postgres@localhost:5432/db")


class Table(Base):
    ...


class Crud(CRUD):
    table = Table


class Service:

    @classmethod
    async def get(cls):
        async with alchemy.connect() as session:
            ...

    @classmethod
    async def create(cls,):
        async with alchemy.connect(transaction=True) as session:
           ...

# testing
import faker

from gadsqlalchemy.testing import Table

fake = faker.Faker()

class Dummy(Table):
    class Meta:
        model = Table

    name = fake.name()

pytest_plugins = [
    "gadsqlalchemy.testing.fixtures",
]