sql_alchemy#

class starlite.plugins.sql_alchemy.SQLAlchemyPlugin#

Bases: PluginProtocol[DeclarativeMeta]

A Plugin for SQLAlchemy.

__init__(config: Optional[SQLAlchemyConfig] = None) → None#

Initialize SQLAlchemyPlugin.

Support (de)serialization and OpenAPI generation for SQLAlchemy ORM types.

Parameters:

config – Optional SQLAlchemyConfig instance. If passed, the plugin will establish a DB connection and hook handlers and dependencies.

on_app_init(app: Starlite) → None#

If config has been passed to the plugin, it will initialize SQLAlchemy and add the dependencies as expected.

Executed on the application’s init process.

Parameters:

app – The Starlite application instance.

Returns:

None

static is_plugin_supported_type(value: Any) → TypeGuard[DeclarativeMeta]#

typing.TypeGuard testing whether values are subclasses of SQLAlchemy’s DeclarativeMeta class.

Parameters:

value – An arbitrary type to test.

Returns:

A boolean typeguard.

static handle_string_type(column_type: Union[String, BINARY, VARBINARY, LargeBinary]) → Type#

Handle the SQLAlchemy String types, including Blob and Binary types.

Parameters:

column_type – The type of the SQLColumn.

Returns:

An appropriate string type

static handle_numeric_type(column_type: Numeric) → Type#

Handle the SQLAlchemy non-int Numeric types.

Parameters:

column_type – The type of the SQLColumn.

Returns:

An appropriate numerical type

handle_list_type(column_type: ARRAY) → Any#

Handle the SQLAlchemy Array type.

Parameters:

column_type – The type of the SQLColumn.

Returns:

An appropriate list type

handle_tuple_type(column_type: TupleType) → Any#

Handle the SQLAlchemy Tuple type.

Parameters:

column_type – The type of the SQLColumn.

Returns:

An appropriate tuple type

static handle_enum(column_type: Union[Enum, ENUM, ENUM]) → Any#

Handle the SQLAlchemy Enum types.

Parameters:

column_type – The type of the SQLColumn.

Returns:

An appropriate enum type

property providers_map: Dict[Type[TypeEngine], Callable[[Union[TypeEngine, Type[TypeEngine]]], Any]]#

Map of SQLAlchemy column types to provider functions.

This method is separated to allow for easy overriding in subclasses.

Returns

A dictionary mapping SQLAlchemy types to callables.

get_pydantic_type(column_type: Any) → Any#

Given a ‘Column.type’ value, return a type supported by pydantic.

Parameters:

column_type – The type of the SQLColumn.

Returns:

A pydantic supported type.

static parse_model(model_class: Type[DeclarativeMeta]) → Mapper#

Validate that the passed in model_class is an SQLAlchemy declarative model, and return a Mapper of it.

Parameters:

model_class – An SQLAlchemy declarative class.

Returns:

A SQLAlchemy Mapper.

to_pydantic_model_class(model_class: Type[DeclarativeMeta], **kwargs: Any) → Type[BaseModel]#

Generate a pydantic model for a given SQLAlchemy declarative table and any nested relations.

Parameters:

  • model_class – An SQLAlchemy declarative class instance.

  • **kwargs – Kwargs to pass to the model.

Returns:

A pydantic model instance.

from_pydantic_model_instance(model_class: Type[DeclarativeMeta], pydantic_model_instance: BaseModel) → Any#

Create an instance of a given model_class using the values stored in the given pydantic_model_instance.

Parameters:

  • model_class – A declarative table class.

  • pydantic_model_instance – A pydantic model instance.

Returns:

A declarative meta table instance.

to_dict(model_instance: DeclarativeMeta) → Dict[str, Any]#

Given a model instance, convert it to a dict of values that can be serialized.

Parameters:

model_instance – An SQLAlchemy declarative table instance.

Returns:

A string keyed dict of values.

from_dict(model_class: Type[DeclarativeMeta], **kwargs: Any) → DeclarativeMeta#

Given a dictionary of kwargs, return an instance of the given model_class.

Parameters:

  • model_class – A declarative table class.

  • **kwargs – Kwargs to instantiate the table with.

Returns:

An instantiated table instance.

class starlite.plugins.sql_alchemy.SQLAlchemyConfig#

Bases: BaseModel

Configuration for SQLAlchemy’s sessionmaker.

For details see: https://docs.sqlalchemy.org/en/14/orm/session_api.html

connection_string: Optional[str]#

Database connection string in one of the formats supported by SQLAlchemy.

Notes

  • For async connections, the connection string must include the correct async prefix. e.g. 'postgresql+asyncpg://...' instead of 'postgresql://', and for sync connections its the opposite.

use_async_engine: bool#

Dictate whether the engine created is an async connection or not.

Notes

  • This option must correlate to the type of connection_string. That is, an async connection string required an async connection and vice versa.

create_async_engine_callable: Callable[[str], AsyncEngine]#

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

create_engine_callable: Callable[[str], Union[Engine, Engine]]#

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

dependency_key: str#

Key to use for the dependency injection of database sessions.

engine_app_state_key: str#

Key under which to store the SQLAlchemy engine in the application State instance.

engine_config: SQLAlchemyEngineConfig#

Configuration for the SQLAlchemy engine.

The configuration options are documented in the SQLAlchemy documentation.

set_json_serializers: bool#

A boolean flag dictating whether to set msgspec based serializer/deserializer functions.

Notes

  • Some databases or some versions of some databases do not have a JSON column type. E.g. some older versions of SQLite for example. In this case this flag should be false or an error will be raised by SQLAlchemy.

session_class: Optional[Union[Type[Session], Type[AsyncSession]]]#

The session class to use.

If not set, the session class will default to sqlalchemy.orm.Session for sync connections and sqlalchemy.ext.asyncio.AsyncSession for async ones.

session_config: SQLAlchemySessionConfig#

Configuration options for the sessionmaker.

The configuration options are documented in the SQLAlchemy documentation.

session_maker_class: Type[SessionMakerTypeProtocol]#

Sessionmaker class to use.

session_maker_app_state_key: str#

Key under which to store the SQLAlchemy sessionmaker in the application State instance.

session_maker_instance: Optional[SessionMakerInstanceProtocol]#

Optional sessionmaker to use.

If set, the plugin will use the provided instance rather than instantiate a sessionmaker.

engine_instance: Optional[Union[Engine, Engine, AsyncEngine]]#

Optional engine to use.

If set, the plugin will use the provided instance rather than instantiate an engine.

before_send_handler: Union[Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Any, Union[HTTPScope, WebSocketScope]], Union[None, Awaitable[None]]], Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Any], Union[None, Awaitable[None]]]]#

Handler to call before the ASGI message is sent.

The handler should handle closing the session stored in the ASGI scope, if its still open, and committing and uncommitted data.

classmethod validate_before_send_handler(value: Union[Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Any, Union[HTTPScope, WebSocketScope]], Union[None, Awaitable[None]]], Callable[[Union[HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent], Any], Union[None, Awaitable[None]]]]) → Any#

Wrap before_send_handler in an AsyncCallable

Parameters:

value – A before send handler callable.

Returns:

An AsyncCallable

classmethod check_connection_string_or_engine_instance(values: Dict[str, Any]) → Dict[str, Any]#

Either connection_string or engine_instance must be specified, and not both.

Parameters:

values – Field values, after validation.

Returns:

Field values.

property engine_config_dict: Dict[str, Any]#

Return the engine configuration as a dict.

Returns:

A string keyed dict of config kwargs for the SQLAlchemy create_engine function.

property engine: Union[Engine, Engine, AsyncEngine]#

Return an engine. If none exists yet, create one.

Returns:

Getter that returns the engine instance used by the plugin.

property session_maker: sessionmaker#

Get a sessionmaker. If none exists yet, create one.

Returns:

Getter that returns the session_maker instance used by the plugin.

create_db_session_dependency(state: State, scope: Scope) → Union[Session, AsyncSession]#

Create a session instance.

Parameters:

  • state – The ‘application.state’ instance.

  • scope – The current connection’s scope.

Returns:

A session instance T.

update_app_state(state: State) → None#

Create a DB engine and stores it in the application state.

Parameters:

state – The ‘application.state’ instance.

Returns:

None

async on_shutdown(state: State) → None#

Disposes of the SQLAlchemy engine.

Parameters:

state – The ‘application.state’ instance.

Returns:

None

config_sql_alchemy_logging(logging_config: Optional[BaseLoggingConfig]) → None#

Add the SQLAlchemy loggers to the logging config.

Notes

Parameters:

logging_config – Logging config.

Returns:

None.

class starlite.plugins.sql_alchemy.SQLAlchemyEngineConfig#

Bases: BaseModel

Configuration for SQLAlchemy’s :class`Engine <sqlalchemy.engine.Engine>`.

For details see: https://docs.sqlalchemy.org/en/14/core/engines.html

class starlite.plugins.sql_alchemy.SQLAlchemySessionConfig#

Bases: BaseModel

Configuration for a SQLAlchemy-Session.