advanced-alchemy

import os from typing import AsyncGenerator from uuid import UUID, uuid4 from advanced_alchemy.base import BaseORMModel from advanced_alchemy.config import AsyncSessionConfig from advanced_alchemy.repository import SQLAlchemyAsyncRepository from sqlalchemy import String from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, mapped_column # Define a database URL (using a simple in-memory SQLite for example) DB_URL = os.environ.get('DB_URL', 'sqlite+aiosqlite:///:memory:') # 1. Define your ORM Model class User(BaseORMModel): __tablename__ = "users" id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4) name: Mapped[str] = mapped_column(String(length=255)) email: Mapped[str] = mapped_column(String(length=255), unique=True) # 2. Configure SQLAlchemy async_session_config = AsyncSessionConfig(url=DB_URL) # 3. Create a Repository for your Model class UserRepository(SQLAlchemyAsyncRepository[User]): model_type = User # 4. Example Usage (e.g., in an async function) async def main(): # Ensure the database tables are created async with async_session_config.get_engine().begin() as conn: await conn.run_sync(BaseORMModel.metadata.create_all) # Get a session maker session_maker = async_session_config.get_session_maker() # Create an instance of the repository with a session async with session_maker() as session: user_repo = UserRepository(session=session) # Create a new user new_user = await user_repo.add(User(name='Alice', email='alice@example.com')) print(f"Created user: {new_user.name} ({new_user.id})") # Fetch all users users = await user_repo.list() print(f"All users: {[u.name for u in users]}") # Update a user updated_user = await user_repo.update(User(id=new_user.id, name='Alicia', email='alicia@example.com')) print(f"Updated user: {updated_user.name}") # Delete a user await user_repo.delete(updated_user.id) deleted_users = await user_repo.list() print(f"Users after deletion: {[u.name for u in deleted_users]}") if __name__ == "__main__": import asyncio asyncio.run(main())