fromadvanced_alchemyimportbase,repository,configfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportMapped,sessionmakerclassUser(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]classUserRepository(repository.SQLAlchemySyncRepository[User]):"""User repository."""model_type=Userdb=config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:",session_config=config.SyncSessionConfig(expire_on_commit=False))# Initializes the database.withdb.get_engine().begin()asconn:User.metadata.create_all(conn)withdb.get_session()asdb_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)forraw_userinbulk_users])db_session.commit()print(f"Created{len(objs)} new objects.")# 2) Select paginated data and total row count.  Pass additional filters as kwargscreated_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.idfornew_objincreated_objs])print(f"Removed{len(deleted_objs)} records out of a total of{total_objs}.")# 4) Let's count the remaining rowsremaining_count=repo.count()print(f"Found{remaining_count} remaining records after delete.")

fromadvanced_alchemyimportbase,repository,filters,service,configfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportMapped,sessionmakerclassUser(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]classUserService(service.SQLAlchemySyncRepositoryService[User]):"""User repository."""classRepo(repository.SQLAlchemySyncRepository[User]):"""User repository."""model_type=Userrepository_type=Repodb=config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:",session_config=config.SyncSessionConfig(expire_on_commit=False))# Initializes the database.withdb.get_engine().begin()asconn:User.metadata.create_all(conn)withdb.get_session()asdb_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 kwargscreated_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.idfornew_objincreated_objs])print(f"Removed{len(deleted_objs)} records out of a total of{total_objs}.")# 4) Let's count the remaining rowsremaining_count=service.count()print(f"Found{remaining_count} remaining records after delete.")

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

fromflaskimportFlaskfromadvanced_alchemy.extensions.flaskimportAdvancedAlchemy,SQLAlchemySyncConfigapp=Flask(__name__)alchemy=AdvancedAlchemy(config=SQLAlchemySyncConfig(connection_string="duckdb:///:memory:"),app=app,)

fromadvanced_alchemy.extensions.fastapiimportAdvancedAlchemy,SQLAlchemyAsyncConfigfromfastapiimportFastAPIapp=FastAPI()alchemy=AdvancedAlchemy(config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"),app=app,)

fromadvanced_alchemy.extensions.starletteimportAdvancedAlchemy,SQLAlchemyAsyncConfigfromstarlette.applicationsimportStarletteapp=Starlette()alchemy=AdvancedAlchemy(config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"),app=app,)

fromsanicimportSanicfromsanic_extimportExtendfromadvanced_alchemy.extensions.sanicimportAdvancedAlchemy,SQLAlchemyAsyncConfigapp=Sanic("AlchemySanicApp")alchemy=AdvancedAlchemy(sqlalchemy_config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"),)Extend.register(alchemy)