SQLAlchemy Plugin#

The SQLAlchemyPlugin provides complete support for working with SQLAlchemy in Litestar applications.

Note

This plugin is only compatible with SQLAlchemy 2.0+.

The SQLAlchemyPlugin combines the functionality of SQLAlchemyInitPlugin and SQLAlchemySerializationPlugin, each of which are examined in detail in the following sections. As such, this section describes a complete example of using the SQLAlchemyPlugin with a Litestar application and a SQLite database.

Or, skip ahead to SQLAlchemy Init Plugin or SQLAlchemy Serialization Plugin to learn more about the individual plugins.

Tip

You can install SQLAlchemy alongside Litestar by running pip install litestar[sqlalchemy].

Example#

SQLAlchemy Async Plugin Example#
 1from __future__ import annotations
 2
 3from typing import TYPE_CHECKING, Sequence
 4
 5from sqlalchemy import select
 6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 7
 8from litestar import Litestar, post
 9from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
10
11if TYPE_CHECKING:
12    from sqlalchemy.ext.asyncio import AsyncSession
13
14
15class Base(DeclarativeBase): ...
16
17
18class TodoItem(Base):
19    __tablename__ = "todo_item"
20    title: Mapped[str] = mapped_column(primary_key=True)
21    done: Mapped[bool]
22
23
24@post("/")
25async def add_item(data: TodoItem, db_session: AsyncSession) -> Sequence[TodoItem]:
26    async with db_session.begin():
27        db_session.add(data)
28    return (await db_session.execute(select(TodoItem))).scalars().all()
29
30
31config = SQLAlchemyAsyncConfig(
32    connection_string="sqlite+aiosqlite:///todo_async.sqlite", create_all=True, metadata=Base.metadata
33)
34plugin = SQLAlchemyPlugin(config=config)
35app = Litestar(route_handlers=[add_item], plugins=[plugin])
SQLAlchemy Async Plugin Example#
 1from __future__ import annotations
 2
 3from typing import TYPE_CHECKING
 4from collections.abc import Sequence
 5
 6from sqlalchemy import select
 7from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 8
 9from litestar import Litestar, post
10from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
11
12if TYPE_CHECKING:
13    from sqlalchemy.ext.asyncio import AsyncSession
14
15
16class Base(DeclarativeBase): ...
17
18
19class TodoItem(Base):
20    __tablename__ = "todo_item"
21    title: Mapped[str] = mapped_column(primary_key=True)
22    done: Mapped[bool]
23
24
25@post("/")
26async def add_item(data: TodoItem, db_session: AsyncSession) -> Sequence[TodoItem]:
27    async with db_session.begin():
28        db_session.add(data)
29    return (await db_session.execute(select(TodoItem))).scalars().all()
30
31
32config = SQLAlchemyAsyncConfig(
33    connection_string="sqlite+aiosqlite:///todo_async.sqlite", create_all=True, metadata=Base.metadata
34)
35plugin = SQLAlchemyPlugin(config=config)
36app = Litestar(route_handlers=[add_item], plugins=[plugin])
SQLAlchemy Sync Plugin Example#
 1from __future__ import annotations
 2
 3from typing import TYPE_CHECKING, Sequence
 4
 5from sqlalchemy import select
 6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 7
 8from litestar import Litestar, post
 9from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemySyncConfig
10
11if TYPE_CHECKING:
12    from sqlalchemy.orm import Session
13
14
15class Base(DeclarativeBase): ...
16
17
18class TodoItem(Base):
19    __tablename__ = "todo_item"
20    title: Mapped[str] = mapped_column(primary_key=True)
21    done: Mapped[bool]
22
23
24@post("/", sync_to_thread=True)
25def add_item(data: TodoItem, db_session: Session) -> Sequence[TodoItem]:
26    with db_session.begin():
27        db_session.add(data)
28    return db_session.execute(select(TodoItem)).scalars().all()
29
30
31config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite", create_all=True, metadata=Base.metadata)
32plugin = SQLAlchemyPlugin(config=config)
33app = Litestar(route_handlers=[add_item], plugins=[plugin])
SQLAlchemy Sync Plugin Example#
 1from __future__ import annotations
 2
 3from typing import TYPE_CHECKING
 4from collections.abc import Sequence
 5
 6from sqlalchemy import select
 7from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 8
 9from litestar import Litestar, post
10from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemySyncConfig
11
12if TYPE_CHECKING:
13    from sqlalchemy.orm import Session
14
15
16class Base(DeclarativeBase): ...
17
18
19class TodoItem(Base):
20    __tablename__ = "todo_item"
21    title: Mapped[str] = mapped_column(primary_key=True)
22    done: Mapped[bool]
23
24
25@post("/", sync_to_thread=True)
26def add_item(data: TodoItem, db_session: Session) -> Sequence[TodoItem]:
27    with db_session.begin():
28        db_session.add(data)
29    return db_session.execute(select(TodoItem)).scalars().all()
30
31
32config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite", create_all=True, metadata=Base.metadata)
33plugin = SQLAlchemyPlugin(config=config)
34app = Litestar(route_handlers=[add_item], plugins=[plugin])

Defining the Database Models#

We start by defining our base model class, and a TodoItem class which extends the base model. The TodoItem class represents a todo item in our SQLite database.

SQLAlchemy Async Plugin Example#
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


@post("/")
SQLAlchemy Sync Plugin Example#
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase): ...


class TodoItem(Base):
    __tablename__ = "todo_item"
    title: Mapped[str] = mapped_column(primary_key=True)
    done: Mapped[bool]


@post("/", sync_to_thread=True)

Setting Up an API Endpoint#

Next, we set up an API endpoint at the root ("/") that allows adding a TodoItem to the SQLite database.

SQLAlchemy Async Plugin Example#
from typing import TYPE_CHECKING, Sequence

from sqlalchemy import select
from litestar import Litestar, post

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession


async def add_item(data: TodoItem, db_session: AsyncSession) -> Sequence[TodoItem]:
    async with db_session.begin():
        db_session.add(data)
    return (await db_session.execute(select(TodoItem))).scalars().all()


config = SQLAlchemyAsyncConfig(
SQLAlchemy Sync Plugin Example#
from typing import TYPE_CHECKING, Sequence

from sqlalchemy import select

if TYPE_CHECKING:
    from sqlalchemy.orm import Session


def add_item(data: TodoItem, db_session: Session) -> Sequence[TodoItem]:
    with db_session.begin():
        db_session.add(data)
    return db_session.execute(select(TodoItem)).scalars().all()


config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite", create_all=True, metadata=Base.metadata)
SQLAlchemy Sync Plugin Example#
from typing import TYPE_CHECKING
from collections.abc import Sequence

from sqlalchemy import select

if TYPE_CHECKING:
    from sqlalchemy.orm import Session


def add_item(data: TodoItem, db_session: Session) -> Sequence[TodoItem]:
    with db_session.begin():
        db_session.add(data)
    return db_session.execute(select(TodoItem)).scalars().all()


config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite", create_all=True, metadata=Base.metadata)

Initializing the Database#

We create a function init_db that we’ll use to initialize the database when the app starts up.

Important

In this example we drop the database before creating it. This is done for the sake of repeatability, and should not be done in production.

SQLAlchemy Async Plugin Example#
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///todo_async.sqlite", create_all=True, metadata=Base.metadata
)
plugin = SQLAlchemyPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin])
SQLAlchemy Sync Plugin Example#
from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemySyncConfig
config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite", create_all=True, metadata=Base.metadata)
plugin = SQLAlchemyPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin])

Setting Up the Plugin and the App#

Finally, we set up the SQLAlchemy Plugin and the Litestar app.

SQLAlchemy Async Plugin Example#
from litestar import Litestar
config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///todo_async.sqlite", create_all=True, metadata=Base.metadata
)
plugin = SQLAlchemyPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin])
SQLAlchemy Sync Plugin Example#
from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemySyncConfig
config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite", create_all=True, metadata=Base.metadata)
plugin = SQLAlchemyPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin])

This configures the app with the plugin, sets up a route handler for adding items, and specifies that the init_db function should be run when the app starts up.

Running the App#

Run the app with the following command:

$ litestar run

You can now add a todo item by sending a POST request to http://localhost:8000 with a JSON body containing the "title" of the todo item.

$ curl -X POST -H "Content-Type: application/json" -d '{"title": "Your Todo Title", "done": false}' http://localhost:8000/