1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
from collections.abc import Sequence
from typing import Union
from litestar.config.app import AppConfig
from litestar.plugins import InitPluginProtocol
from advanced_alchemy.extensions.litestar.plugins import _slots_base
from advanced_alchemy.extensions.litestar.plugins.init import (
EngineConfig,
SQLAlchemyAsyncConfig,
SQLAlchemyInitPlugin,
SQLAlchemySyncConfig,
)
from advanced_alchemy.extensions.litestar.plugins.serialization import SQLAlchemySerializationPlugin
class SQLAlchemyPlugin(InitPluginProtocol, _slots_base.SlotsBase):
"""A plugin that provides SQLAlchemy integration."""
def __init__(
self,
config: Union[
SQLAlchemyAsyncConfig, SQLAlchemySyncConfig, Sequence[Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]]
],
) -> None:
"""Initialize ``SQLAlchemyPlugin``.
Args:
config: configure DB connection and hook handlers and dependencies.
"""
self._config = config if isinstance(config, Sequence) else [config]
@property
def config(
self,
) -> Sequence[Union[SQLAlchemyAsyncConfig, SQLAlchemySyncConfig]]:
return self._config
def on_app_init(self, app_config: AppConfig) -> AppConfig:
"""Configure application for use with SQLAlchemy.
Args:
app_config: The :class:`AppConfig <.config.app.AppConfig>` instance.
Returns:
The :class:`AppConfig <.config.app.AppConfig>` instance.
"""
app_config.plugins.extend([SQLAlchemyInitPlugin(config=self._config), SQLAlchemySerializationPlugin()])
return app_config
__all__ = (
"EngineConfig",
"SQLAlchemyAsyncConfig",
"SQLAlchemyInitPlugin",
"SQLAlchemyPlugin",
"SQLAlchemySerializationPlugin",
"SQLAlchemySyncConfig",
)
|
