from advanced_alchemy import base, repository, config from sqlalchemy import create_engine from sqlalchemy.orm import Mapped, sessionmaker class User(base.UUIDBase): # you can optionally override the generated table name by manually setting it. __tablename__ = "user_account" # type: ignore[assignment] email: Mapped[str] name: Mapped[str] class UserRepository(repository.SQLAlchemySyncRepository[User]): """User repository.""" model_type = User db = config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:", session_config=config.SyncSessionConfig(expire_on_commit=False)) # Initializes the database. with db.get_engine().begin() as conn: User.metadata.create_all(conn) with db.get_session() as db_session: repo = UserRepository(session=db_session) # 1) Create multiple users with `add_many` bulk_users = [ {"email": 'cody@litestar.dev', 'name': 'Cody'}, {"email": 'janek@litestar.dev', 'name': 'Janek'}, {"email": 'peter@litestar.dev', 'name': 'Peter'}, {"email": 'jacob@litestar.dev', 'name': 'Jacob'} ] objs = repo.add_many([User(**raw_user) for raw_user in bulk_users]) db_session.commit() print(f"Created {len(objs)} new objects.") # 2) Select paginated data and total row count. Pass additional filters as kwargs created_objs, total_objs = repo.list_and_count(LimitOffset(limit=10, offset=0), name="Cody") print(f"Selected {len(created_objs)} records out of a total of {total_objs}.") # 3) Let's remove the batch of records selected. deleted_objs = repo.delete_many([new_obj.id for new_obj in created_objs]) print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.") # 4) Let's count the remaining rows remaining_count = repo.count() print(f"Found {remaining_count} remaining records after delete.")

from advanced_alchemy import base, repository, filters, service, config from sqlalchemy import create_engine from sqlalchemy.orm import Mapped, sessionmaker class User(base.UUIDBase): # you can optionally override the generated table name by manually setting it. __tablename__ = "user_account" # type: ignore[assignment] email: Mapped[str] name: Mapped[str] class UserService(service.SQLAlchemySyncRepositoryService[User]): """User repository.""" class Repo(repository.SQLAlchemySyncRepository[User]): """User repository.""" model_type = User repository_type = Repo db = config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:", session_config=config.SyncSessionConfig(expire_on_commit=False)) # Initializes the database. with db.get_engine().begin() as conn: User.metadata.create_all(conn) with db.get_session() as db_session: service = UserService(session=db_session) # 1) Create multiple users with `add_many` objs = service.create_many([ {"email": 'cody@litestar.dev', 'name': 'Cody'}, {"email": 'janek@litestar.dev', 'name': 'Janek'}, {"email": 'peter@litestar.dev', 'name': 'Peter'}, {"email": 'jacob@litestar.dev', 'name': 'Jacob'} ]) print(objs) print(f"Created {len(objs)} new objects.") # 2) Select paginated data and total row count. Pass additional filters as kwargs created_objs, total_objs = service.list_and_count(LimitOffset(limit=10, offset=0), name="Cody") print(f"Selected {len(created_objs)} records out of a total of {total_objs}.") # 3) Let's remove the batch of records selected. deleted_objs = service.delete_many([new_obj.id for new_obj in created_objs]) print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.") # 4) Let's count the remaining rows remaining_count = service.count() print(f"Found {remaining_count} remaining records after delete.")

from litestar import Litestar from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemyAsyncConfig # alternately... # from advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig, SQLAlchemyPlugin alchemy = SQLAlchemyPlugin( config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), ) app = Litestar(plugins=[alchemy])

from flask import Flask from advanced_alchemy.extensions.flask import AdvancedAlchemy, SQLAlchemySyncConfig app = Flask(__name__) alchemy = AdvancedAlchemy( config=SQLAlchemySyncConfig(connection_string="duckdb:///:memory:"), app=app, )

from advanced_alchemy.extensions.fastapi import AdvancedAlchemy, SQLAlchemyAsyncConfig from fastapi import FastAPI app = FastAPI() alchemy = AdvancedAlchemy( config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app, )

from advanced_alchemy.extensions.starlette import AdvancedAlchemy, SQLAlchemyAsyncConfig from starlette.applications import Starlette app = Starlette() alchemy = AdvancedAlchemy( config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app, )

from sanic import Sanic from sanic_ext import Extend from advanced_alchemy.extensions.sanic import AdvancedAlchemy, SQLAlchemyAsyncConfig app = Sanic("AlchemySanicApp") alchemy = AdvancedAlchemy( sqlalchemy_config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), ) Extend.register(alchemy)