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#
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy import select
6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
7
8from litestar import Litestar, post
9from litestar.contrib.sqlalchemy.plugins import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
10
11if TYPE_CHECKING:
12
13 from sqlalchemy.ext.asyncio import AsyncSession
14
15
16class Base(DeclarativeBase):
17 ...
18
19
20class TodoItem(Base):
21 __tablename__ = "todo_item"
22 title: Mapped[str] = mapped_column(primary_key=True)
23 done: Mapped[bool]
24
25
26@post("/")
27async def add_item(data: TodoItem, db_session: AsyncSession) -> list[TodoItem]:
28 async with db_session.begin():
29 db_session.add(data)
30 return (await db_session.execute(select(TodoItem))).scalars().all()
31
32
33async def init_db(app: Litestar) -> None:
34 async with app.state.db_engine.begin() as conn:
35 await conn.run_sync(Base.metadata.drop_all)
36 await conn.run_sync(Base.metadata.create_all)
37
38
39config = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///todo_async.sqlite")
40plugin = SQLAlchemyPlugin(config=config)
41app = Litestar(route_handlers=[add_item], plugins=[plugin], on_startup=[init_db])
1from __future__ import annotations
2
3from typing import TYPE_CHECKING
4
5from sqlalchemy import select
6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
7
8from litestar import Litestar, post
9from litestar.contrib.sqlalchemy.plugins import SQLAlchemyPlugin, SQLAlchemySyncConfig
10
11if TYPE_CHECKING:
12
13 from sqlalchemy.orm import Session
14
15
16class Base(DeclarativeBase):
17 ...
18
19
20class TodoItem(Base):
21 __tablename__ = "todo_item"
22 title: Mapped[str] = mapped_column(primary_key=True)
23 done: Mapped[bool]
24
25
26@post("/", sync_to_thread=True)
27def add_item(data: TodoItem, db_session: Session) -> list[TodoItem]:
28 with db_session.begin():
29 db_session.add(data)
30 return db_session.execute(select(TodoItem)).scalars().all()
31
32
33def init_db(app: Litestar) -> None:
34 Base.metadata.drop_all(app.state.db_engine)
35 Base.metadata.create_all(app.state.db_engine)
36
37
38config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite")
39plugin = SQLAlchemyPlugin(config=config)
40app = Litestar(route_handlers=[add_item], plugins=[plugin], on_startup=[init_db])
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.
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]
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]
Setting Up an API Endpoint#
Next, we set up an API endpoint at the root ("/"
) that allows adding a TodoItem
to the SQLite database.
from typing import TYPE_CHECKING
from sqlalchemy import select
from litestar import post
if TYPE_CHECKING:
from typing import List
from sqlalchemy.ext.asyncio import AsyncSession
@post("/")
async def add_item(data: TodoItem, db_session: AsyncSession) -> List[TodoItem]:
async with db_session.begin():
db_session.add(data)
return (await db_session.execute(select(TodoItem))).scalars().all()
from typing import TYPE_CHECKING
from sqlalchemy import select
from litestar import post
if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession
@post("/")
async def add_item(data: TodoItem, db_session: AsyncSession) -> list[TodoItem]:
async with db_session.begin():
db_session.add(data)
return (await db_session.execute(select(TodoItem))).scalars().all()
from typing import TYPE_CHECKING
from sqlalchemy import select
from litestar import post
if TYPE_CHECKING:
from typing import List
from sqlalchemy.orm import Session
@post("/", sync_to_thread=True)
def add_item(data: TodoItem, db_session: Session) -> List[TodoItem]:
with db_session.begin():
db_session.add(data)
return db_session.execute(select(TodoItem)).scalars().all()
from typing import TYPE_CHECKING
from sqlalchemy import select
from litestar import post
if TYPE_CHECKING:
from sqlalchemy.orm import Session
@post("/", sync_to_thread=True)
def add_item(data: TodoItem, db_session: Session) -> list[TodoItem]:
with db_session.begin():
db_session.add(data)
return db_session.execute(select(TodoItem)).scalars().all()
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.
from litestar import Litestar
async def init_db(app: Litestar) -> None:
async with app.state.db_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
from litestar import Litestar
def init_db(app: Litestar) -> None:
Base.metadata.drop_all(app.state.db_engine)
Base.metadata.create_all(app.state.db_engine)
Setting Up the Plugin and the App#
Finally, we set up the SQLAlchemy Plugin and the Litestar app.
from litestar import Litestar
from litestar.contrib.sqlalchemy.plugins import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
config = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///todo_async.sqlite")
plugin = SQLAlchemyPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin], on_startup=[init_db])
from litestar import Litestar
from litestar.contrib.sqlalchemy.plugins import SQLAlchemyPlugin, SQLAlchemySyncConfig
config = SQLAlchemySyncConfig(connection_string="sqlite:///todo_sync.sqlite")
plugin = SQLAlchemyPlugin(config=config)
app = Litestar(route_handlers=[add_item], plugins=[plugin], on_startup=[init_db])
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/