sync

Sync SQLAlchemy configuration module.

class advanced_alchemy.config.sync.AlembicSyncConfig

Bases: GenericAlembicConfig

Configuration for Alembic’s synchronous migrations.

For details see: https://alembic.sqlalchemy.org/en/latest/api/config.html

__init__(script_config='alembic.ini', version_table_name='alembic_versions', version_table_schema=None, script_location='migrations', toml_file=None, user_module_prefix='sa.', render_as_batch=True, compare_type=False, template_path='/home/runner/work/advanced-alchemy/advanced-alchemy/advanced_alchemy/alembic/templates')

class advanced_alchemy.config.sync.SQLAlchemySyncConfig

Bases: GenericSQLAlchemyConfig[Engine, Session, sessionmaker[Session]]

Synchronous SQLAlchemy Configuration.

Note

The alembic configuration options are documented in the Alembic documentation.

Example

Basic sync configuration:

config = SQLAlchemySyncConfig(
    connection_string="postgresql://user:pass@localhost/db",
)

Configuration with read/write routing:

from advanced_alchemy.config.routing import RoutingConfig

config = SQLAlchemySyncConfig(
    routing_config=RoutingConfig(
        primary_connection_string="postgresql://user:pass@primary/db",
        read_replicas=["postgresql://user:pass@replica/db"],
    ),
)

create_engine_callable(**kwargs)

Overloads:

• url (Union[str, URL]), connect_args (Dict[Any, Any]), convert_unicode (bool), creator (Union[_CreatorFnType, _CreatorWRecFnType]), echo (_EchoFlagType), echo_pool (_EchoFlagType), enable_from_linting (bool), execution_options (_ExecuteOptions), future (Literal[True]), hide_parameters (bool), implicit_returning (Literal[True]), insertmanyvalues_page_size (int), isolation_level (IsolationLevel), json_deserializer (Callable[…, Any]), json_serializer (Callable[…, Any]), label_length (Optional[int]), logging_name (str), max_identifier_length (Optional[int]), max_overflow (int), module (Optional[Any]), paramstyle (Optional[_ParamStyle]), pool (Optional[Pool]), poolclass (Optional[Type[Pool]]), pool_logging_name (str), pool_pre_ping (bool), pool_size (int), pool_recycle (int), pool_reset_on_return (Optional[_ResetStyleArgType]), pool_timeout (float), pool_use_lifo (bool), plugins (List[str]), query_cache_size (int), use_insertmanyvalues (bool), kwargs (Any) → Engine

• url (Union[str, URL]), kwargs (Any) → Engine

Parameters:

• url (Union[str, _url.URL])

• kwargs (Any)

Return type:

Engine

Callable that creates an Engine instance or instance of its subclass.

session_config: SyncSessionConfig

Configuration options for the sessionmaker.

session_maker_class

Sessionmaker class to use.

alias of sessionmaker

alembic_config: AlembicSyncConfig

Configuration for the SQLAlchemy Alembic migrations.

The configuration options are documented in the Alembic documentation.

routing_config: RoutingConfig | None = None

Optional read/write routing configuration.

When provided, enables automatic routing of read operations to replicas and write operations to the primary database.

Note

When using routing_config, do not set connection_string. The primary connection is specified in the routing config.

__init__(create_engine_callable=<function create_engine>, session_config=<factory>, session_maker_class=<class 'sqlalchemy.orm.session.sessionmaker'>, connection_string=None, engine_config=<factory>, session_maker=None, engine_instance=None, create_all=False, metadata=None, bind_key=None, enable_touch_updated_timestamp_listener=True, enable_file_object_listener=True, file_object_raise_on_error=True, alembic_config=<factory>, routing_config=None)

create_session_maker()

Get a session maker.

If routing is configured, returns a routing-aware session maker. Otherwise, returns a standard session maker.

Returns:

A callable that creates session instances.

Return type:

Callable[[], Session]

get_session()

Get a session context manager.

Yields:

Generator[sqlalchemy.orm.Session, None, None] – A context manager yielding an active SQLAlchemy Session.

Return type:

Generator[Session, None, None]

Examples

Using the session context manager:

>>> with config.get_session() as session:
...     session.execute(...)

class advanced_alchemy.config.sync.SyncSessionConfig

Bases: GenericSessionConfig[Connection, Engine, Session]

Configuration for synchronous SQLAlchemy sessions.

__init__(autobegin=<class 'advanced_alchemy.utils.dataclass.Empty'>, autoflush=<class 'advanced_alchemy.utils.dataclass.Empty'>, bind=<class 'advanced_alchemy.utils.dataclass.Empty'>, binds=<class 'advanced_alchemy.utils.dataclass.Empty'>, class_=<class 'advanced_alchemy.utils.dataclass.Empty'>, expire_on_commit=<class 'advanced_alchemy.utils.dataclass.Empty'>, info=<class 'advanced_alchemy.utils.dataclass.Empty'>, join_transaction_mode=<class 'advanced_alchemy.utils.dataclass.Empty'>, query_cls=<class 'advanced_alchemy.utils.dataclass.Empty'>, twophase=<class 'advanced_alchemy.utils.dataclass.Empty'>)