[Python] How to use migration commands with Litestar?

Discussão em 'Python' iniciado por Stack, Abril 20, 2025.

  1. Stack

    Stack Membro Participativo

    I am trying to use migration commands with Litestar, but I keep encountering the error: No such command 'database'.

    I have configured my project with pyproject.toml, main.py, and models.user.py as shown below. I also use advanced-alchemy for database integration and Alembic for migrations. However, I cannot find a way to run database migration commands through Litestar.

    Here are the relevant files:

    pyproject.toml:

    [tool.poetry]
    name = "test-task"
    version = "0.1.0"
    description = "A Python web service using LiteStar and PostgreSQL."
    license = "MIT"
    package-mode = false

    [tool.poetry.dependencies]
    python = "^3.12"
    python-dotenv = "^1"
    litestar = { extras = ["standard"], version = "^2" }
    litestar-granian = "^0"
    litestar-asyncpg = "^0"
    pydantic = "^1.10"
    advanced-alchemy = "^0.20"
    msgspec = "^0.18.6"
    asyncpg = "^0"

    [tool.poetry.dev-dependencies]
    pytest = "^7.0"
    pytest-asyncio = "^0.20"
    alembic = "^1.11.1"

    [tool.poetry.scripts]
    advanced-alchemy = "advanced_alchemy.cli:cli"

    [build-system]
    requires = ["poetry-core>=1.0.0"]
    build-backend = "poetry.core.masonry.api"

    [tool.litestar.app]
    path = "app.py"
    attribute = "app"

    [tool.litestar.database]
    connection_string = "postgresql+asyncpg://postgres:postgres@db:5432/tasks_example"
    metadata_module = "models"
    base_class = "Base"

    [tool.alembic]
    script_location = "migrations"
    sqlalchemy.url = "postgresql+asyncpg://postgres:postgres@db:5432/tasks_example"


    main.py:

    from litestar import Litestar
    from litestar.openapi.config import OpenAPIConfig
    from app.controllers.user_controller import UserController
    from config import DATABASE_URL

    from litestar.plugins.sqlalchemy import (
    AsyncSessionConfig,
    SQLAlchemyAsyncConfig,
    SQLAlchemyPlugin,
    )

    session_config = AsyncSessionConfig(expire_on_commit=False)
    sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string=DATABASE_URL,
    before_send_handler="autocommit",
    session_config=session_config,
    create_all=True,
    )
    alchemy = SQLAlchemyPlugin(config=sqlalchemy_config)

    openapi_config = OpenAPIConfig(
    title="Users API",
    version="1.0.0",
    description="API",
    )

    app = Litestar(
    route_handlers=[UserController],
    openapi_config=openapi_config,
    plugins=[alchemy],
    )

    if __name__ == "__main__":
    app.run()


    models.tasks:

    from sqlalchemy import Column, BigInteger, String, TIMESTAMP, func
    from advanced_alchemy.extensions.litestar import (
    base
    )

    class BaseModel(base.UUIDBase):
    __abstract__ = True

    id = Column(BigInteger, primary_key=True, autoincrement=True)
    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now(), nullable=False)
    updated_at = Column(TIMESTAMP(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)

    class Task(BaseModel):
    __tablename__ = 'task'

    name = Column(String, nullable=False)

    Continue reading...

Compartilhe esta Página