# SQLAlchemy bind manager
[](https://pypi.org/project/sqlalchemy-bind-manager/)
[](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#beta)
[](https://github.com/febus982/sqlalchemy-bind-manager/actions/workflows/python-tests.yml)
[](https://codeclimate.com/github/febus982/sqlalchemy-bind-manager/maintainability)
[](https://codeclimate.com/github/febus982/sqlalchemy-bind-manager/test_coverage)
[](https://mypy-lang.org/)
[](https://github.com/psf/black)
[](https://github.com/charliermarsh/ruff)
[](https://github.com/PyCQA/bandit)
This package provides an easy way to configure and use SQLAlchemy engines and sessions
without depending on frameworks.
It is composed by two main components:
* A manager class for SQLAlchemy engine and session configuration
* A repository/unit-of-work pattern implementation for model retrieval and persistence
## Installation
```bash
pip install sqlalchemy-bind-manager
```
## Components maturity
[//]: # (https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md)
* [](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#beta) **SQLAlchemy manager:** Implementation is mostly finalised, needs testing in production.
* [](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#beta) **Repository:** Implementation is mostly finalised, needs testing in production.
* [](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#experimental) **Unit of work:** The implementation is working but limited to repositories using the same engine. Distributed transactions across different engines are not yet supported.
## Documentation
The complete documentation can be found [here](https://febus982.github.io/sqlalchemy-bind-manager)
## SQLAlchemy manager
Initialise the manager providing an instance of `SQLAlchemyConfig`
```python
from sqlalchemy_bind_manager import SQLAlchemyConfig, SQLAlchemyBindManager
config = SQLAlchemyConfig(
engine_url="sqlite:///./sqlite.db",
engine_options=dict(connect_args={"check_same_thread": False}, echo=True),
session_options=dict(expire_on_commit=False),
)
sa_manager = SQLAlchemyBindManager(config)
```
🚨 NOTE: Using global variables is not thread-safe, please read the [Documentation](https://febus982.github.io/sqlalchemy-bind-manager/manager/session/#note-on-multithreaded-applications) if your application uses multi-threading.
The `engine_url` and `engine_options` dictionaries accept the same parameters as SQLAlchemy [create_engine()](https://docs.sqlalchemy.org/en/14/core/engines.html#sqlalchemy.create_engine)
The `session_options` dictionary accepts the same parameters as SQLALchemy [sessionmaker()](https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.sessionmaker)
The `SQLAlchemyBindManager` provides some helper methods for common operations:
* `get_bind`: returns a `SQLAlchemyBind` or `SQLAlchemyAsyncBind` object
* `get_session`: returns a `Session` object, which works also as a context manager
* `get_mapper`: returns the mapper associated with the bind
Example:
```python
bind = sa_manager.get_bind()
class MyModel(bind.declarative_base):
pass
# Persist an object
o = MyModel()
o.name = "John"
with sa_manager.get_session() as session:
session.add(o)
session.commit()
```
[Imperative model declaration](https://febus982.github.io/sqlalchemy-bind-manager/manager/models/) is also supported.
### Multiple database binds
`SQLAlchemyBindManager` accepts also multiple databases configuration, provided as a dictionary. The dictionary keys are used as a reference name for each bind.
```python
from sqlalchemy_bind_manager import SQLAlchemyConfig, SQLAlchemyBindManager
config = {
"default": SQLAlchemyConfig(
engine_url="sqlite:///./sqlite.db",
engine_options=dict(connect_args={"check_same_thread": False}, echo=True),
session_options=dict(expire_on_commit=False),
),
"secondary": SQLAlchemyConfig(
engine_url="sqlite:///./secondary.db",
engine_options=dict(connect_args={"check_same_thread": False}, echo=True),
session_options=dict(expire_on_commit=False),
),
}
sa_manager = SQLAlchemyBindManager(config)
```
All the `SQLAlchemyBindManager` helper methods accept the `bind_name` optional parameter:
* `get_bind(bind_name="default")`: returns a `SQLAlchemyBind` or `SQLAlchemyAsyncBind` object
* `get_session(bind_name="default")`: returns a `Session` or `AsyncSession` object, which works also as a context manager
* `get_mapper(bind_name="default")`: returns the mapper associated with the bind
### Asynchronous database binds
Is it possible to supply configurations for asyncio supported engines.
```python
config = SQLAlchemyAsyncConfig(
engine_url="postgresql+asyncpg://scott:tiger@localhost/test",
)
```
This will make sure we have an `AsyncEngine` and an `AsyncSession` are initialised, as an asynchronous context manager.
```python
async with sa_manager.get_session() as session:
session.add(o)
await session.commit()
```
Note that async implementation has several differences from the sync one, make sure
to check [SQLAlchemy asyncio documentation](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
## Repository / Unit of work
The `SQLAlchemyRepository` and `SQLAlchemyAsyncRepository` class can be used directly or by extending them.
```python
from sqlalchemy_bind_manager.repository import SQLAlchemyRepository
class MyModel(declarative_base):
pass
# Direct usage
repo_instance = SQLAlchemyRepository(sqlalchemy_bind_manager.get_bind(), model_class=MyModel)
class ModelRepository(SQLAlchemyRepository[MyModel]):
_model = MyModel
def _some_custom_method_implemented(self):
...
# Extended class usage
extended_repo_instance = ModelRepository(sqlalchemy_bind_manager.get_bind())
```
The repository classes provides methods for common use case:
* `get`: Retrieve a model by primary key
* `save`: Persist a model
* `save_many`: Persist multiple models in a single transaction
* `delete`: Delete a model
* `find`: Search for a list of models (basically an adapter for SELECT queries)
* `paginated_find`: Search for a list of models, with pagination support
* `cursor_paginated_find`: Search for a list of models, with cursor based pagination support
### Use the Unit Of Work to share a session among multiple repositories
It is possible we need to run several operations in a single database transaction. While a single
repository provide by itself an isolated session for single operations, we have to use a different
approach for multiple operations.
We can use the `UnitOfWork` or the `AsyncUnitOfWork` class to provide a shared session to
be used for repository operations, **assumed the same bind is used for all the repositories**.
```python
class MyRepo(SQLAlchemyRepository):
_model = MyModel
bind = sa_manager.get_bind()
uow = UnitOfWork(bind)
uow.register_repository("repo_a", MyRepo)
uow.register_repository("repo_b", SQLAlchemyRepository, MyOtherModel)
with uow.transaction():
uow.repository("repo_a").save(some_model)
uow.repository("repo_b").save(some_other_model)
```
Raw data
{
"_id": null,
"home_page": "https://febus982.github.io/sqlalchemy-bind-manager",
"name": "sqlalchemy-bind-manager",
"maintainer": null,
"docs_url": null,
"requires_python": "<3.14,>=3.9",
"maintainer_email": null,
"keywords": "sqlalchemy, config, manager",
"author": "Federico Busetti",
"author_email": "729029+febus982@users.noreply.github.com",
"download_url": "https://files.pythonhosted.org/packages/69/44/19a0ba44a615728de26ba30bd9a971fbe0866591ce855ecd78d6e064092e/sqlalchemy_bind_manager-0.11.0.tar.gz",
"platform": null,
"description": "# SQLAlchemy bind manager\n\n[](https://pypi.org/project/sqlalchemy-bind-manager/)\n[](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#beta)\n\n[](https://github.com/febus982/sqlalchemy-bind-manager/actions/workflows/python-tests.yml)\n[](https://codeclimate.com/github/febus982/sqlalchemy-bind-manager/maintainability)\n[](https://codeclimate.com/github/febus982/sqlalchemy-bind-manager/test_coverage)\n\n[](https://mypy-lang.org/)\n[](https://github.com/psf/black)\n[](https://github.com/charliermarsh/ruff)\n[](https://github.com/PyCQA/bandit)\n\nThis package provides an easy way to configure and use SQLAlchemy engines and sessions\nwithout depending on frameworks.\n\nIt is composed by two main components:\n\n* A manager class for SQLAlchemy engine and session configuration\n* A repository/unit-of-work pattern implementation for model retrieval and persistence\n\n## Installation\n\n```bash\npip install sqlalchemy-bind-manager\n```\n\n## Components maturity\n\n[//]: # (https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md)\n* [](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#beta) **SQLAlchemy manager:** Implementation is mostly finalised, needs testing in production.\n* [](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#beta) **Repository:** Implementation is mostly finalised, needs testing in production.\n* [](https://github.com/mkenney/software-guides/blob/master/STABILITY-BADGES.md#experimental) **Unit of work:** The implementation is working but limited to repositories using the same engine. Distributed transactions across different engines are not yet supported.\n\n## Documentation\n\nThe complete documentation can be found [here](https://febus982.github.io/sqlalchemy-bind-manager)\n\n## SQLAlchemy manager \n\nInitialise the manager providing an instance of `SQLAlchemyConfig`\n\n```python\nfrom sqlalchemy_bind_manager import SQLAlchemyConfig, SQLAlchemyBindManager\n\nconfig = SQLAlchemyConfig(\n engine_url=\"sqlite:///./sqlite.db\",\n engine_options=dict(connect_args={\"check_same_thread\": False}, echo=True),\n session_options=dict(expire_on_commit=False),\n)\n\nsa_manager = SQLAlchemyBindManager(config)\n```\n\n\ud83d\udea8 NOTE: Using global variables is not thread-safe, please read the [Documentation](https://febus982.github.io/sqlalchemy-bind-manager/manager/session/#note-on-multithreaded-applications) if your application uses multi-threading.\n\nThe `engine_url` and `engine_options` dictionaries accept the same parameters as SQLAlchemy [create_engine()](https://docs.sqlalchemy.org/en/14/core/engines.html#sqlalchemy.create_engine)\n\nThe `session_options` dictionary accepts the same parameters as SQLALchemy [sessionmaker()](https://docs.sqlalchemy.org/en/14/orm/session_api.html#sqlalchemy.orm.sessionmaker)\n\nThe `SQLAlchemyBindManager` provides some helper methods for common operations:\n\n* `get_bind`: returns a `SQLAlchemyBind` or `SQLAlchemyAsyncBind` object\n* `get_session`: returns a `Session` object, which works also as a context manager\n* `get_mapper`: returns the mapper associated with the bind\n\nExample:\n\n```python\nbind = sa_manager.get_bind()\n\n\nclass MyModel(bind.declarative_base):\n pass\n\n\n# Persist an object\no = MyModel()\no.name = \"John\"\nwith sa_manager.get_session() as session:\n session.add(o)\n session.commit()\n```\n\n[Imperative model declaration](https://febus982.github.io/sqlalchemy-bind-manager/manager/models/) is also supported.\n\n### Multiple database binds\n\n`SQLAlchemyBindManager` accepts also multiple databases configuration, provided as a dictionary. The dictionary keys are used as a reference name for each bind.\n\n```python\nfrom sqlalchemy_bind_manager import SQLAlchemyConfig, SQLAlchemyBindManager\n\nconfig = {\n \"default\": SQLAlchemyConfig(\n engine_url=\"sqlite:///./sqlite.db\",\n engine_options=dict(connect_args={\"check_same_thread\": False}, echo=True),\n session_options=dict(expire_on_commit=False),\n ),\n \"secondary\": SQLAlchemyConfig(\n engine_url=\"sqlite:///./secondary.db\",\n engine_options=dict(connect_args={\"check_same_thread\": False}, echo=True),\n session_options=dict(expire_on_commit=False),\n ),\n}\n\nsa_manager = SQLAlchemyBindManager(config)\n```\n\nAll the `SQLAlchemyBindManager` helper methods accept the `bind_name` optional parameter:\n\n* `get_bind(bind_name=\"default\")`: returns a `SQLAlchemyBind` or `SQLAlchemyAsyncBind` object\n* `get_session(bind_name=\"default\")`: returns a `Session` or `AsyncSession` object, which works also as a context manager\n* `get_mapper(bind_name=\"default\")`: returns the mapper associated with the bind\n\n### Asynchronous database binds\n\nIs it possible to supply configurations for asyncio supported engines.\n\n```python\nconfig = SQLAlchemyAsyncConfig(\n engine_url=\"postgresql+asyncpg://scott:tiger@localhost/test\",\n)\n```\n\nThis will make sure we have an `AsyncEngine` and an `AsyncSession` are initialised, as an asynchronous context manager.\n\n```python\nasync with sa_manager.get_session() as session:\n session.add(o)\n await session.commit()\n```\n\nNote that async implementation has several differences from the sync one, make sure\nto check [SQLAlchemy asyncio documentation](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)\n\n## Repository / Unit of work\n\nThe `SQLAlchemyRepository` and `SQLAlchemyAsyncRepository` class can be used directly or by extending them.\n\n```python\nfrom sqlalchemy_bind_manager.repository import SQLAlchemyRepository\n\n\nclass MyModel(declarative_base):\n pass\n\n# Direct usage\nrepo_instance = SQLAlchemyRepository(sqlalchemy_bind_manager.get_bind(), model_class=MyModel)\n\nclass ModelRepository(SQLAlchemyRepository[MyModel]):\n _model = MyModel\n \n def _some_custom_method_implemented(self):\n ...\n\n# Extended class usage\nextended_repo_instance = ModelRepository(sqlalchemy_bind_manager.get_bind())\n```\n\nThe repository classes provides methods for common use case:\n\n* `get`: Retrieve a model by primary key\n* `save`: Persist a model\n* `save_many`: Persist multiple models in a single transaction\n* `delete`: Delete a model\n* `find`: Search for a list of models (basically an adapter for SELECT queries)\n* `paginated_find`: Search for a list of models, with pagination support\n* `cursor_paginated_find`: Search for a list of models, with cursor based pagination support\n\n### Use the Unit Of Work to share a session among multiple repositories\n\nIt is possible we need to run several operations in a single database transaction. While a single\nrepository provide by itself an isolated session for single operations, we have to use a different\napproach for multiple operations.\n\nWe can use the `UnitOfWork` or the `AsyncUnitOfWork` class to provide a shared session to\nbe used for repository operations, **assumed the same bind is used for all the repositories**.\n\n```python\nclass MyRepo(SQLAlchemyRepository):\n _model = MyModel\n\nbind = sa_manager.get_bind()\nuow = UnitOfWork(bind)\nuow.register_repository(\"repo_a\", MyRepo)\nuow.register_repository(\"repo_b\", SQLAlchemyRepository, MyOtherModel)\n\nwith uow.transaction():\n uow.repository(\"repo_a\").save(some_model)\n uow.repository(\"repo_b\").save(some_other_model)\n```\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A manager to easily handle multiple SQLAlchemy configurations",
"version": "0.11.0",
"project_urls": {
"Homepage": "https://febus982.github.io/sqlalchemy-bind-manager",
"Repository": "https://github.com/febus982/sqlalchemy-bind-manager"
},
"split_keywords": [
"sqlalchemy",
" config",
" manager"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b65efb76f6b877a075eb6a302bfbbffd7cb8bae069533b7fafeea0441da79240",
"md5": "4d55041b304b77a06438e9718de753c7",
"sha256": "5592084f6e6cfd834960f375e1ca691722f062feaad521095f87d72e0d594141"
},
"downloads": -1,
"filename": "sqlalchemy_bind_manager-0.11.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "4d55041b304b77a06438e9718de753c7",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<3.14,>=3.9",
"size": 30195,
"upload_time": "2024-12-15T15:33:14",
"upload_time_iso_8601": "2024-12-15T15:33:14.828660Z",
"url": "https://files.pythonhosted.org/packages/b6/5e/fb76f6b877a075eb6a302bfbbffd7cb8bae069533b7fafeea0441da79240/sqlalchemy_bind_manager-0.11.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "694419a0ba44a615728de26ba30bd9a971fbe0866591ce855ecd78d6e064092e",
"md5": "9008e6201f67fac66d85fa173d3953a6",
"sha256": "f84767d80a3aca5cb8b29de0c38d95368799dcc74a3b5c9d0ba303757d4300e8"
},
"downloads": -1,
"filename": "sqlalchemy_bind_manager-0.11.0.tar.gz",
"has_sig": false,
"md5_digest": "9008e6201f67fac66d85fa173d3953a6",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<3.14,>=3.9",
"size": 18455,
"upload_time": "2024-12-15T15:33:18",
"upload_time_iso_8601": "2024-12-15T15:33:18.493014Z",
"url": "https://files.pythonhosted.org/packages/69/44/19a0ba44a615728de26ba30bd9a971fbe0866591ce855ecd78d6e064092e/sqlalchemy_bind_manager-0.11.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-15 15:33:18",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "febus982",
"github_project": "sqlalchemy-bind-manager",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "sqlalchemy-bind-manager"
}
JSON
