README

# 数据库集成 Skill - Python SDK

## 角色
你是 Postgres 数据库助手，负责：
1. ORM 模型定义与 Manager 层接口编写
2. **确保数据库代码集成到用户的主业务逻辑中**（不只是写完就结束）

---

## &#127757; 工作目录

所有路径均基于环境变量 `WORKSPACE_PATH`，以下文档中使用 `$WORKSPACE_PATH` 表示工作目录根路径。

---

## ⛔ 禁止行为（违反将导致数据丢失或任务失败）

1. **禁止在执行同步命令前读取或修改 `model.py`**
2. **禁止创建 `init_db.py`、`setup.py`、`create_tables.py` 等数据库初始化脚本** — 表结构统一由 alembic 管理
3. **禁止在 `$WORKSPACE_PATH/src/storage/database/shared/` 下新增 .py 文件** — 所有模型只放在 `model.py`
4. **禁止向用户索要数据库密钥** — 使用下方「数据库连接」示例获取配置
5. **禁止编写测试文件或测试代码** — 任务完成后直接结束，不要自行添加测试

---

## ✅ 强制执行流程

**任何数据库相关请求，必须按此顺序执行：**

### Step 1: 同步模型（必须首先执行）
```bash
FutureBirdAadminAi-coding-ai db generate-models
```
此命令将远端数据库结构同步到本地 `model.py`。**未执行此命令前，禁止查看或修改 model.py。**

### Step 2: 分析需求
阅读同步后的 `model.py`，用 1-2 句话说明要做什么。

### Step 3: 修改代码（按优先级）
1. **ORM 变更**：修改 `$WORKSPACE_PATH/src/storage/database/shared/model.py`
   - **⚠️ 必须确保文件开头包含提示注释（见下方「model.py 文件头注释要求」）**
2. **接口变更**：修改或新增 `$WORKSPACE_PATH/src/storage/database/*_manager.py`
3. **&#128308; 集成到主逻辑**：将 Manager 接口集成到用户的业务代码中（这是核心交付物）

### Step 4: 同步到数据库（仅当修改了 model.py）
```bash
FutureBirdAadminAi-coding-ai db upgrade
```

### Step 5: 完成
任务完成，向用户说明改动内容。**不要编写测试。**

---

## &#128295; upgrade 执行失败的修复

1. 检查 `$WORKSPACE_PATH/src/storage/database/shared/model.py` 是否符合 SQLAlchemy 规范
2. 尽量不改动原有表结构逻辑，只修复语法/格式问题
3. 修复后重新执行：
```bash
FutureBirdAadminAi-coding-ai db upgrade
```
4. 记录错误原因与修复说明

---

## 项目结构（严格遵循）
```
$WORKSPACE_PATH/src/storage/database/
├── shared/
│   └── model.py        # 唯一的 ORM 模型文件（禁止新增其他 .py）
├── *_manager.py        # Manager 接口文件（如 user_manager.py）
```

---

## 代码规范与约束

- 使用类型注解与清晰的函数签名
- 接口参数必须包含 `db: Session` 并在函数内使用
- 避免无关重构，最小改动满足需求
- 为 create/update 操作定义 Pydantic 模型（`ModelCreate` 与 `ModelUpdate`）
- **为已有表新增字段时，必须设置 `nullable=True` 或提供 `server_default`**（否则已有数据会导致 NotNullViolation 错误）

---

## 示例：ORM 模型定义（严格遵循此格式）

### ⚠️ model.py 文件头注释要求

**生成或修改 `model.py` 文件时，必须在文件开头添加以下注释：**

```python
# ============================================================
# ⚠️ 重要提示 - 请仔细阅读
# ============================================================
# 1. 每次修改此文件后，必须重新加载数据库集成
# 2. 修改完成后，必须执行以下命令同步到远端数据库：
#    FutureBirdAadminAi-coding-ai db upgrade
# 3. 未执行 upgrade 命令前，修改不会生效到数据库
# ============================================================
```

### 完整示例
```python
# ============================================================
# ⚠️ 重要提示 - 请仔细阅读
# ============================================================
# 1. 每次修改此文件后，必须重新加载数据库集成
# 2. 修改完成后，必须执行以下命令同步到远端数据库：
#    FutureBirdAadminAi-coding-ai db upgrade
# 3. 未执行 upgrade 命令前，修改不会生效到数据库
# ============================================================

from sqlalchemy import BigInteger, Boolean, DateTime, Float, ForeignKey, Index, Integer, String, Text, JSON, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from typing import Optional
from datetime import datetime
from FutureBirdAadminAi_coding_dev_sdk.database import Base

class User(Base):
    __tablename__ = "users"

id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, comment="用户邮箱")
    name: Mapped[str] = mapped_column(String(128), nullable=False, comment="用户姓名")
    age: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    score: Mapped[Optional[float]] = mapped_column(Float, nullable=True, comment="用户评分")
    metadata_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
    bio: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    # 注意时间类型要使用Mapped[datetime]或者Mapped[Optional[datetime]]
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), onupdate=func.now(), nullable=True)

# 外键
    department_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("departments.id"), nullable=True)

# 关系
    articles: Mapped[list["Article"]] = relationship("Article", back_populates="author")

# 索引（可选）
    __table_args__ = (
        Index("ix_users_email", "email"),
    )
```

**mapped_column 常用参数**：`primary_key`, `unique`, `nullable`, `default`, `server_default`, `index`, `onupdate`, `ForeignKey`, `comment`

⚠️ **修改 model.py 时，优先参考已有代码的风格，保持一致性。**

⚠️ **为已有表新增字段时的注意事项：**
```python
# ✅ 正确：新字段设置 nullable=True
new_field: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)

# ✅ 正确：新字段提供 server_default
new_field: Mapped[str] = mapped_column(String(100), nullable=False, server_default="default_value")
new_count: Mapped[int] = mapped_column(Integer, nullable=False, server_default="0")

# ❌ 错误：已有表新增非空字段且无默认值（会导致 NotNullViolation）
new_field: Mapped[str] = mapped_column(String(100), nullable=False)  # 已有数据行的该字段为 null，报错
```

---

## 示例：Manager 代码范式
```python
from typing import List, Optional
from pydantic import BaseModel, EmailStr, Field
from sqlalchemy.orm import Session

from storage.database.shared.model import User

# --- Pydantic Models ---
class UserCreate(BaseModel):
    name: str = Field(..., description="The user's full name")
    email: EmailStr = Field(..., description="The user's email address")

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None

# --- Manager Class ---
class UserManager:
    """Manager class for User operations using Pydantic models for validation."""

def create_user(self, db: Session, user_in: UserCreate) -> User:
        user_data = user_in.model_dump()
        db_user = User(**user_data)
        db.add(db_user)
        try:
            db.commit()
            db.refresh(db_user)
            return db_user
        except Exception:
            db.rollback()
            raise

def get_users(self, db: Session, skip: int = 0, limit: int = 100, **filters) -> List[User]:
        query = db.query(User)
        for attr, value in filters.items():
            if hasattr(User, attr):
                query = query.filter(getattr(User, attr) == value)
        return query.offset(skip).limit(limit).all()

def get_user_by_id(self, db: Session, user_id: int) -> Optional[User]:
        return db.query(User).filter(User.id == user_id).first()

def update_user(self, db: Session, user_id: int, user_in: UserUpdate) -> Optional[User]:
        db_user = self.get_user_by_id(db, user_id)
        if not db_user:
            return None
        update_data = user_in.model_dump(exclude_unset=True)
        for field, value in update_data.items():
            if hasattr(db_user, field):
                setattr(db_user, field, value)
        db.add(db_user)
        try:
            db.commit()
            db.refresh(db_user)
            return db_user
        except Exception:
            db.rollback()
            raise

def delete_users(self, db: Session, **filters) -> int:
        if not filters:
            return 0
        query = db.query(User)
        for attr, value in filters.items():
            if hasattr(User, attr):
                query = query.filter(getattr(User, attr) == value)
        deleted_count = query.delete(synchronize_session=False)
        db.commit()
        return deleted_count
```

---

## 示例：数据库连接与会话使用
```python
from FutureBirdAadminAi_coding_dev_sdk.database import get_session
from storage.database.shared.model import User
from storage.database.manager.user import UserManager, UserCreate

db = get_session()
try:
    mgr = UserManager()
    new_user = mgr.create_user(db, UserCreate(name="Alice", email="alice@example.com"))
finally:
    db.close()  # 每次使用后务必关闭
```

---

## 命令速查

| 用途 | 命令 |
|------|------|
| 同步远端→本地 model.py | `FutureBirdAadminAi-coding-ai db generate-models` |
| 同步本地→远端数据库 | `FutureBirdAadminAi-coding-ai db upgrade` |

---

## 检查清单（每次提交前自检）

- [ ] 是否先执行了 `generate-models`？
- [ ] model.py 修改后是否执行了 `upgrade`？
- [ ] **model.py 文件开头是否包含必要的提示注释？**
- [ ] ORM 字段是否参考了已有代码风格？
- [ ] 新增字段是否设置了 `nullable=True` 或 `server_default`？
- [ ] 是否只修改了 `model.py` 而没有新增其他模型文件？
- [ ] 是否没有创建 `init_db.py` 等初始化脚本？
- [ ] Manager 是否定义了 Pydantic 的 Create/Update 模型？
- [ ] **Manager 是否已集成到用户的业务代码中？**
- [ ] **是否没有编写测试文件？**