数据库表

Agno 模板预配置了 SqlAlchemy 和 alembic 来管理数据库。添加表的通用工作流程是：

1. 将表定义添加到 db/tables 目录。
2. 在 db/tables/__init__.py 文件中导入表类。
3. 创建数据库迁移。
4. 运行数据库迁移。

表定义

我们来创建一个 UsersTable，将以下代码复制到 db/tables/user.py

db/tables/user.py

from datetime import datetime
from typing import Optional

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.sql.expression import text
from sqlalchemy.types import BigInteger, DateTime, String

from db.tables.base import Base


class UsersTable(Base):
    """存储用户数据的表。"""

    __tablename__ = "dim_users"

    id_user: Mapped[int] = mapped_column(
        BigInteger, primary_key=True, autoincrement=True, nullable=False, index=True
    )
    email: Mapped[str] = mapped_column(String)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=text("now()")
    )
    updated_at: Mapped[Optional[datetime]] = mapped_column(
        DateTime(timezone=True), onupdate=text("now()")
    )

更新 db/tables/__init__.py 文件：

db/tables/__init__.py

from db.tables.base import Base
from db.tables.user import UsersTable

创建数据库修订

在开发容器中运行 alembic 命令来创建数据库迁移：

docker exec -it ai-api alembic -c db/alembic.ini revision --autogenerate -m "Initialize DB"

迁移开发数据库

运行 alembic 命令来迁移开发数据库：

docker exec -it ai-api alembic -c db/alembic.ini upgrade head

可选：添加测试用户

现在我们来添加一个测试用户。将以下代码复制到 db/tables/test_add_user.py

db/tables/test_add_user.py

from typing import Optional
from sqlalchemy.orm import Session

from db.session import SessionLocal
from db.tables.user import UsersTable
from utils.log import logger


def create_user(db_session: Session, email: str) -> UsersTable:
    """创建新用户。"""
    new_user = UsersTable(email=email)
    db_session.add(new_user)
    return new_user


def get_user(db_session: Session, email: str) -> Optional[UsersTable]:
    """通过电子邮件获取用户。"""
    return db_session.query(UsersTable).filter(UsersTable.email == email).first()


if __name__ == "__main__":
    test_user_email = "[e-mail]"
    with SessionLocal() as sess, sess.begin():
        logger.info(f"正在创建用户: {test_user_email}")
        create_user(db_session=sess, email=test_user_email)
        logger.info(f"正在获取用户: {test_user_email}")
        user = get_user(db_session=sess, email=test_user_email)
        if user:
            logger.info(f"用户已创建: {user.id_user}")
        else:
            logger.info(f"用户未找到: {test_user_email}")

运行脚本以添加测试用户：

docker exec -it ai-api python db/tables/test_add_user.py

迁移生产数据库

我们建议通过设置环境变量 MIGRATE_DB = True 并重新启动生产服务来迁移生产数据库。这将从入口点脚本在容器启动时运行 alembic -c db/alembic.ini upgrade head。

更新 workspace/prd_resources.py 文件

workspace/prd_resources.py

...
# -*- 构建容器环境
container_env = {
    ...
    # 启动时使用 alembic 迁移数据库
    "MIGRATE_DB": ws_settings.prd_db_enabled,
}
...

更新 ECS 任务定义

由于我们更新了环境变量，我们需要更新任务定义：

terminal

ag ws patch --env prd --infra aws --name td

更新 ECS 服务

更新任务定义后，重新部署生产应用程序：

terminal

ag ws patch --env prd --infra aws --name service

手动迁移生产数据库

另一种方法是 SSH 进入生产容器以手动运行迁移。您的 ECS 任务已启用 SSH 访问。运行 alembic 命令来迁移生产数据库：

ECS_CLUSTER=ai-app-prd-cluster
TASK_ARN=$(aws ecs list-tasks --cluster ai-app-prd-cluster --query "taskArns[0]" --output text)
CONTAINER_NAME=ai-api-prd

aws ecs execute-command --cluster $ECS_CLUSTER \
    --task $TASK_ARN \
    --container $CONTAINER_NAME \
    --interactive \
    --command "alembic -c db/alembic.ini upgrade head"

---

migrations 目录是如何创建的

这些命令已经运行过，并在此描述以供完整性参考

migrations 目录是使用以下命令创建的：

docker exec -it ai-api cd db && alembic init migrations

• 运行上述命令后，应会创建 db/migrations 目录。
• 更新 alembic.ini
  • 将 script_location = db/migrations 设置为
  • 取消注释 [post_write_hooks] 中的 black hook
• 按照 此链接 更新 db/migrations/env.py 文件
• 将以下函数添加到 configure 中，以仅包含 target_metadata 中的表

db/migrations/env.py

# -*- 只包含 target_metadata 中的表
def include_name(name, type_, parent_names):
    if type_ == "table":
        return name in target_metadata.tables
    else:
        return True
...