SQLAlchemy Plugin

Starlite comes with built-in support for SQLAlchemy via the SQLAlchemyPlugin.

Features

• Managed sessions (sync and async) including dependency injection

• Automatic serialization of SQLAlchemy models powered pydantic

• Data validation based on SQLAlchemy models powered pydantic

See also

The following examples use SQLAlchemy’s “2.0 Style” introduced in SQLAlchemy 1.4.

If you are unfamiliar with it, you can find a comprehensive migration guide in SQLAlchemy’s documentation here, and a handy table comparing the ORM usage

Attention

The SQLAlchemyPlugin supports only mapped classes. Tables are currently not supported since they are not easy to convert to pydantic models.

Basic Use

You can simply pass an instance of SQLAlchemyPlugin without passing config to the Starlite constructor. This will extend support for serialization, deserialization and DTO creation for SQLAlchemy declarative models:

Async

Python 3.8+

sqlalchemy_plugin.py

from typing import Optional

from sqlalchemy import Column, Float, Integer, String, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, declarative_base

from starlite import DTOFactory, HTTPException, Starlite, get, post
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

Base = declarative_base()

sqlalchemy_config = SQLAlchemyConfig(
    connection_string="sqlite+aiosqlite:///test.sqlite", dependency_key="async_session"
)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
dto_factory = DTOFactory(plugins=[sqlalchemy_plugin])


class Company(Base):  # pyright: ignore
    __tablename__ = "company"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    worth: Mapped[float] = Column(Float)


CreateCompanyDTO = dto_factory("CreateCompanyDTO", Company, exclude=["id"])


async def on_startup() -> None:
    """Initialize the database."""
    async with sqlalchemy_config.engine.begin() as conn:  # type: ignore
        await conn.run_sync(Base.metadata.create_all)  # pyright: ignore


@post(path="/companies")
async def create_company(
    data: CreateCompanyDTO,  # type: ignore[valid-type]
    async_session: AsyncSession,
) -> Company:
    """Create a new company and return it."""
    company: Company = data.to_model_instance()  # type: ignore[attr-defined]
    async_session.add(company)
    await async_session.commit()
    return company


@get(path="/companies/{company_id:int}")
async def get_company(company_id: int, async_session: AsyncSession) -> Company:
    """Get a company by its ID and return it.

    If a company with that ID does not exist, return a 404 response
    """
    result = await async_session.scalars(select(Company).where(Company.id == company_id))
    company: Optional[Company] = result.one_or_none()
    if not company:
        raise HTTPException(detail=f"Company with ID {company_id} not found", status_code=HTTP_404_NOT_FOUND)
    return company


app = Starlite(
    route_handlers=[create_company, get_company],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Python 3.10+

sqlalchemy_plugin.py

from sqlalchemy import Column, Float, Integer, String, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, declarative_base

from starlite import DTOFactory, HTTPException, Starlite, get, post
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

Base = declarative_base()

sqlalchemy_config = SQLAlchemyConfig(
    connection_string="sqlite+aiosqlite:///test.sqlite", dependency_key="async_session"
)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
dto_factory = DTOFactory(plugins=[sqlalchemy_plugin])


class Company(Base):  # pyright: ignore
    __tablename__ = "company"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    worth: Mapped[float] = Column(Float)


CreateCompanyDTO = dto_factory("CreateCompanyDTO", Company, exclude=["id"])


async def on_startup() -> None:
    """Initialize the database."""
    async with sqlalchemy_config.engine.begin() as conn:  # type: ignore
        await conn.run_sync(Base.metadata.create_all)  # pyright: ignore


@post(path="/companies")
async def create_company(
    data: CreateCompanyDTO,  # type: ignore[valid-type]
    async_session: AsyncSession,
) -> Company:
    """Create a new company and return it."""
    company: Company = data.to_model_instance()  # type: ignore[attr-defined]
    async_session.add(company)
    await async_session.commit()
    return company


@get(path="/companies/{company_id:int}")
async def get_company(company_id: int, async_session: AsyncSession) -> Company:
    """Get a company by its ID and return it.

    If a company with that ID does not exist, return a 404 response
    """
    result = await async_session.scalars(select(Company).where(Company.id == company_id))
    company: Company | None = result.one_or_none()
    if not company:
        raise HTTPException(detail=f"Company with ID {company_id} not found", status_code=HTTP_404_NOT_FOUND)
    return company


app = Starlite(
    route_handlers=[create_company, get_company],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Sync

Python 3.8+

sqlalchemy_plugin.py

from typing import Optional

from sqlalchemy import Column, Float, Integer, String, select
from sqlalchemy.orm import Mapped, Session, declarative_base

from starlite import DTOFactory, HTTPException, Starlite, get, post
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

Base = declarative_base()

sqlalchemy_config = SQLAlchemyConfig(connection_string="sqlite+pysqlite:///test.sqlite", use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
dto_factory = DTOFactory(plugins=[sqlalchemy_plugin])


class Company(Base):  # pyright: ignore
    __tablename__ = "company"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    worth: Mapped[float] = Column(Float)


CreateCompanyDTO = dto_factory("CreateCompanyDTO", Company, exclude=["id"])


def on_startup() -> None:
    """Initialize the database."""
    Base.metadata.create_all(sqlalchemy_config.engine)  # type: ignore


@post(path="/companies")
def create_company(
    data: CreateCompanyDTO,  # type: ignore[valid-type]
    db_session: Session,
) -> Company:
    """Create a new company and return it."""
    company: Company = data.to_model_instance()  # type: ignore[attr-defined]
    db_session.add(company)
    db_session.commit()
    return company


@get(path="/companies/{company_id:int}")
def get_company(company_id: str, db_session: Session) -> Company:
    """Get a company by its ID and return it.

    If a company with that ID does not exist, return a 404 response
    """
    company: Optional[Company] = db_session.scalars(select(Company).where(Company.id == company_id)).one_or_none()
    if not company:
        raise HTTPException(detail=f"Company with ID {company_id} not found", status_code=HTTP_404_NOT_FOUND)
    return company


app = Starlite(
    route_handlers=[create_company, get_company],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Python 3.10+

sqlalchemy_plugin.py

from sqlalchemy import Column, Float, Integer, String, select
from sqlalchemy.orm import Mapped, Session, declarative_base

from starlite import DTOFactory, HTTPException, Starlite, get, post
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

Base = declarative_base()

sqlalchemy_config = SQLAlchemyConfig(connection_string="sqlite+pysqlite:///test.sqlite", use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)
dto_factory = DTOFactory(plugins=[sqlalchemy_plugin])


class Company(Base):  # pyright: ignore
    __tablename__ = "company"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    worth: Mapped[float] = Column(Float)


CreateCompanyDTO = dto_factory("CreateCompanyDTO", Company, exclude=["id"])


def on_startup() -> None:
    """Initialize the database."""
    Base.metadata.create_all(sqlalchemy_config.engine)  # type: ignore


@post(path="/companies")
def create_company(
    data: CreateCompanyDTO,  # type: ignore[valid-type]
    db_session: Session,
) -> Company:
    """Create a new company and return it."""
    company: Company = data.to_model_instance()  # type: ignore[attr-defined]
    db_session.add(company)
    db_session.commit()
    return company


@get(path="/companies/{company_id:int}")
def get_company(company_id: str, db_session: Session) -> Company:
    """Get a company by its ID and return it.

    If a company with that ID does not exist, return a 404 response
    """
    company: Company | None = db_session.scalars(select(Company).where(Company.id == company_id)).one_or_none()
    if not company:
        raise HTTPException(detail=f"Company with ID {company_id} not found", status_code=HTTP_404_NOT_FOUND)
    return company


app = Starlite(
    route_handlers=[create_company, get_company],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Using imperative mappings

Imperative mappings are supported as well, just make sure to use a mapped class instead of the table itself

company_table = Table(
    "company",
    Base.registry.metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("worth", Float),
)


class Company:
    pass


Base.registry.map_imperatively(Company, company_table)

Relationships

Attention

Currently only to-one relationships are supported because of the way the SQLAlchemy plugin handles relationships. Since it recursively traverses relationships, a cyclic reference will result in an endless loop. To prevent this, these relationships will be type as typing.Any in the pydantic model Relationships are typed as typing.Optional in the pydantic model by default so sending incomplete models won’t cause any issues.

Simple relationships

Simple relationships can be handled by the plugin automatically:

Python 3.8+

sqlalchemy_relationships.py

from typing import Optional

from sqlalchemy import Column, Float, ForeignKey, Integer, String, create_engine, select
from sqlalchemy.orm import Mapped, Session, declarative_base, relationship

from starlite import HTTPException, Starlite, get
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

engine = create_engine("sqlite+pysqlite://")
sqlalchemy_config = SQLAlchemyConfig(engine_instance=engine, use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)

Base = declarative_base()


class Company(Base):  # pyright: ignore
    __tablename__ = "company"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    worth: Mapped[float] = Column(Float)


class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    company_id: Mapped[int] = Column(Integer, ForeignKey("company.id"))
    company: Mapped[Company] = relationship("Company", uselist=False)


async def on_startup() -> None:
    """Initialize the database."""
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        peter = User(id=1, name="Peter", company=Company(name="Peter Co.", worth=0.0))
        session.add(peter)
        session.commit()


@get(path="/user/{user_id:int}")
def get_user(user_id: int, db_session: Session) -> User:
    """Get a user by its ID and return it.

    If a user with that ID does not exist, return a 404 response
    """
    user: Optional[User] = db_session.scalars(select(User).where(User.id == user_id)).one_or_none()
    if not user:
        raise HTTPException(detail=f"User with ID {user} not found", status_code=HTTP_404_NOT_FOUND)
    return user


app = Starlite(
    route_handlers=[get_user],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Python 3.10+

sqlalchemy_relationships.py

from sqlalchemy import Column, Float, ForeignKey, Integer, String, create_engine, select
from sqlalchemy.orm import Mapped, Session, declarative_base, relationship

from starlite import HTTPException, Starlite, get
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

engine = create_engine("sqlite+pysqlite://")
sqlalchemy_config = SQLAlchemyConfig(engine_instance=engine, use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)

Base = declarative_base()


class Company(Base):  # pyright: ignore
    __tablename__ = "company"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    worth: Mapped[float] = Column(Float)


class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    company_id: Mapped[int] = Column(Integer, ForeignKey("company.id"))
    company: Mapped[Company] = relationship("Company", uselist=False)


async def on_startup() -> None:
    """Initialize the database."""
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        peter = User(id=1, name="Peter", company=Company(name="Peter Co.", worth=0.0))
        session.add(peter)
        session.commit()


@get(path="/user/{user_id:int}")
def get_user(user_id: int, db_session: Session) -> User:
    """Get a user by its ID and return it.

    If a user with that ID does not exist, return a 404 response
    """
    user: User | None = db_session.scalars(select(User).where(User.id == user_id)).one_or_none()
    if not user:
        raise HTTPException(detail=f"User with ID {user} not found", status_code=HTTP_404_NOT_FOUND)
    return user


app = Starlite(
    route_handlers=[get_user],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Example

Run the above with uvicorn sqlalchemy_relationships:app, navigate your browser to http://127.0.0.0:8000/user/1 and you will see:

{
  "id": 1,
  "name": "Peter",
  "company_id": 1,
  "company": {
    "id": 1,
    "name": "Peter Co.",
    "worth": 0
  }
}

To-Many relationships and circular references

For to-many relationships or those that contain circular references you need to define the pydantic models yourself:

Python 3.8+

sqlalchemy_relationships_to_many

from typing import List, Optional

from pydantic import BaseModel
from sqlalchemy import Column, ForeignKey, Integer, String, create_engine, select
from sqlalchemy.orm import Mapped, Session, declarative_base, relationship

from starlite import HTTPException, Starlite, get
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

engine = create_engine("sqlite+pysqlite:///test.sqlite")
sqlalchemy_config = SQLAlchemyConfig(engine_instance=engine, use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)

Base = declarative_base()


class Pet(Base):
    __tablename__ = "pet"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    owner: Mapped["User"] = relationship("User", back_populates="pets", uselist=True)
    owner_id: Mapped[int] = Column(ForeignKey("user.id"))


class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String, default="moishe")
    pets: Mapped[List[Pet]] = relationship("Pet", uselist=True, back_populates="owner")


class PetModel(BaseModel):
    class Config:
        orm_mode = True

    id: int
    name: str


class UserModel(BaseModel):
    class Config:
        orm_mode = True

    id: int
    name: str
    pets: List[PetModel]


def on_startup() -> None:
    """Initialize the database."""
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        peter = User(id=1, name="Peter", pets=[Pet(id=1, name="Paul")])
        session.add(peter)
        session.commit()


@get(path="/user/{user_id:int}")
def get_user(user_id: int, db_session: Session) -> UserModel:
    """Get a user by its ID and return it.

    If a user with that ID does not exist, return a 404 response
    """
    user: Optional[User] = db_session.scalars(select(User).where(User.id == user_id)).one_or_none()
    if not user:
        raise HTTPException(detail=f"User with ID {user} not found", status_code=HTTP_404_NOT_FOUND)
    return UserModel.from_orm(user)


app = Starlite(
    route_handlers=[get_user],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Python 3.9+

sqlalchemy_relationships_to_many

from typing import Optional

from pydantic import BaseModel
from sqlalchemy import Column, ForeignKey, Integer, String, create_engine, select
from sqlalchemy.orm import Mapped, Session, declarative_base, relationship

from starlite import HTTPException, Starlite, get
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

engine = create_engine("sqlite+pysqlite:///test.sqlite")
sqlalchemy_config = SQLAlchemyConfig(engine_instance=engine, use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)

Base = declarative_base()


class Pet(Base):
    __tablename__ = "pet"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    owner: Mapped["User"] = relationship("User", back_populates="pets", uselist=True)
    owner_id: Mapped[int] = Column(ForeignKey("user.id"))


class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String, default="moishe")
    pets: Mapped[list[Pet]] = relationship("Pet", uselist=True, back_populates="owner")


class PetModel(BaseModel):
    class Config:
        orm_mode = True

    id: int
    name: str


class UserModel(BaseModel):
    class Config:
        orm_mode = True

    id: int
    name: str
    pets: list[PetModel]


def on_startup() -> None:
    """Initialize the database."""
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        peter = User(id=1, name="Peter", pets=[Pet(id=1, name="Paul")])
        session.add(peter)
        session.commit()


@get(path="/user/{user_id:int}")
def get_user(user_id: int, db_session: Session) -> UserModel:
    """Get a user by its ID and return it.

    If a user with that ID does not exist, return a 404 response
    """
    user: Optional[User] = db_session.scalars(select(User).where(User.id == user_id)).one_or_none()
    if not user:
        raise HTTPException(detail=f"User with ID {user} not found", status_code=HTTP_404_NOT_FOUND)
    return UserModel.from_orm(user)


app = Starlite(
    route_handlers=[get_user],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Python 3.10+

sqlalchemy_relationships_to_many

from pydantic import BaseModel
from sqlalchemy import Column, ForeignKey, Integer, String, create_engine, select
from sqlalchemy.orm import Mapped, Session, declarative_base, relationship

from starlite import HTTPException, Starlite, get
from starlite.plugins.sql_alchemy import SQLAlchemyConfig, SQLAlchemyPlugin
from starlite.status_codes import HTTP_404_NOT_FOUND

engine = create_engine("sqlite+pysqlite:///test.sqlite")
sqlalchemy_config = SQLAlchemyConfig(engine_instance=engine, use_async_engine=False)
sqlalchemy_plugin = SQLAlchemyPlugin(config=sqlalchemy_config)

Base = declarative_base()


class Pet(Base):
    __tablename__ = "pet"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String)
    owner: Mapped["User"] = relationship("User", back_populates="pets", uselist=True)
    owner_id: Mapped[int] = Column(ForeignKey("user.id"))


class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = Column(Integer, primary_key=True)
    name: Mapped[str] = Column(String, default="moishe")
    pets: Mapped[list[Pet]] = relationship("Pet", uselist=True, back_populates="owner")


class PetModel(BaseModel):
    class Config:
        orm_mode = True

    id: int
    name: str


class UserModel(BaseModel):
    class Config:
        orm_mode = True

    id: int
    name: str
    pets: list[PetModel]


def on_startup() -> None:
    """Initialize the database."""
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        peter = User(id=1, name="Peter", pets=[Pet(id=1, name="Paul")])
        session.add(peter)
        session.commit()


@get(path="/user/{user_id:int}")
def get_user(user_id: int, db_session: Session) -> UserModel:
    """Get a user by its ID and return it.

    If a user with that ID does not exist, return a 404 response
    """
    user: User | None = db_session.scalars(select(User).where(User.id == user_id)).one_or_none()
    if not user:
        raise HTTPException(detail=f"User with ID {user} not found", status_code=HTTP_404_NOT_FOUND)
    return UserModel.from_orm(user)


app = Starlite(
    route_handlers=[get_user],
    on_startup=[on_startup],
    plugins=[sqlalchemy_plugin],
)

Example

Run the above with uvicorn sqlalchemy_relationships_to_many:app, navigate your browser to http://127.0.0.0:8000/user/1 and you will see:

{
  "id": 1,
  "name": "Peter",
  "pets": [
    {
      "id": 1,
      "name": "Paul"
    }
  ]
}

Configuration

You can configure the Plugin using the SQLAlchemyConfig object.