Back to Details

database-management

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Database Management Skill

数据库管理技能

When to Activate

激活场景

Activate this skill when:
  • Setting up database functionality
  • Creating database schemas
  • Implementing data persistence
  • Writing database queries
  • Working with SQLite or any database
IMPORTANT: This is MANDATORY for all projects requiring database functionality.
在以下场景激活该技能：
  • 搭建数据库功能
  • 创建数据库模式
  • 实现数据持久化
  • 编写数据库查询语句
  • 处理SQLite或其他数据库相关工作
重要提示：所有需要数据库功能的项目均强制使用该技能。

Core Architecture

核心架构

Principles

设计原则

  1. SQLite Only: Use SQLite as default database
  2. Single Interface: All database operations through 
    database.py
  3. Complete SQL Isolation: All SQL statements in 
    database.py
  4. Function-Based: Simple, reusable function interface
  1. 仅使用SQLite：默认使用SQLite作为数据库
  2. 单一接口：所有数据库操作均通过
    database.py
    完成
  3. 完全SQL隔离：所有SQL语句都应放在
    database.py
    中
  4. 基于函数：采用简洁、可复用的函数式接口

File Structure

文件结构

project/
├── database.py         # ALL SQL lives here
├── app.py              # Uses database functions (no SQL!)
└── tests/
    └── test_database.py
project/
├── database.py         # ALL SQL lives here
├── app.py              # Uses database functions (no SQL!)
└── tests/
    └── test_database.py

Standard Interface Pattern

标准接口模式

python
undefined
python
undefined

database.py - All database code lives here

database.py - All database code lives here

import sqlite3 from typing import List, Dict, Optional, Any
DB_PATH = "app.db"
def get_connection(db_path: str = DB_PATH) -> sqlite3.Connection: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row return conn
def init_db(db_path: str = DB_PATH) -> None: """Initialize database with schema.""" conn = get_connection(db_path) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, email TEXT UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() conn.close()
def db_query(query: str, params: tuple = ()) -> List[Dict[str, Any]]: """Execute SELECT query and return results.""" conn = get_connection() cursor = conn.cursor() cursor.execute(query, params) rows = cursor.fetchall() conn.close() return [dict(row) for row in rows]
def db_execute(query: str, params: tuple = ()) -> int: """Execute INSERT/UPDATE/DELETE and return affected rows.""" conn = get_connection() cursor = conn.cursor() cursor.execute(query, params) conn.commit() affected = cursor.rowcount conn.close() return affected
def db_insert(query: str, params: tuple = ()) -> int: """Execute INSERT and return last row ID.""" conn = get_connection() cursor = conn.cursor() cursor.execute(query, params) conn.commit() last_id = cursor.lastrowid conn.close() return last_id
undefined
import sqlite3 from typing import List, Dict, Optional, Any
DB_PATH = "app.db"
def get_connection(db_path: str = DB_PATH) -> sqlite3.Connection: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row return conn
def init_db(db_path: str = DB_PATH) -> None: """Initialize database with schema.""" conn = get_connection(db_path) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, email TEXT UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() conn.close()
def db_query(query: str, params: tuple = ()) -> List[Dict[str, Any]]: """Execute SELECT query and return results.""" conn = get_connection() cursor = conn.cursor() cursor.execute(query, params) rows = cursor.fetchall() conn.close() return [dict(row) for row in rows]
def db_execute(query: str, params: tuple = ()) -> int: """Execute INSERT/UPDATE/DELETE and return affected rows.""" conn = get_connection() cursor = conn.cursor() cursor.execute(query, params) conn.commit() affected = cursor.rowcount conn.close() return affected
def db_insert(query: str, params: tuple = ()) -> int: """Execute INSERT and return last row ID.""" conn = get_connection() cursor = conn.cursor() cursor.execute(query, params) conn.commit() last_id = cursor.lastrowid conn.close() return last_id
undefined

Domain-Specific Functions

领域专属函数

python
undefined
python
undefined

Add to database.py - Clean API for application code

Add to database.py - Clean API for application code

def get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]: """Get user by ID.""" results = db_query("SELECT * FROM users WHERE id = ?", (user_id,)) return results[0] if results else None
def get_user_by_email(email: str) -> Optional[Dict[str, Any]]: """Get user by email.""" results = db_query("SELECT * FROM users WHERE email = ?", (email,)) return results[0] if results else None
def create_user(username: str, email: str) -> int: """Create new user and return ID.""" return db_insert( "INSERT INTO users (username, email) VALUES (?, ?)", (username, email) )
def delete_user(user_id: int) -> bool: """Delete user by ID.""" return db_execute("DELETE FROM users WHERE id = ?", (user_id,)) > 0
undefined
def get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]: """Get user by ID.""" results = db_query("SELECT * FROM users WHERE id = ?", (user_id,)) return results[0] if results else None
def get_user_by_email(email: str) -> Optional[Dict[str, Any]]: """Get user by email.""" results = db_query("SELECT * FROM users WHERE email = ?", (email,)) return results[0] if results else None
def create_user(username: str, email: str) -> int: """Create new user and return ID.""" return db_insert( "INSERT INTO users (username, email) VALUES (?, ?)", (username, email) )
def delete_user(user_id: int) -> bool: """Delete user by ID.""" return db_execute("DELETE FROM users WHERE id = ?", (user_id,)) > 0
undefined

Application Usage

应用层使用示例

python
undefined
python
undefined

app.py - NO SQL HERE!

app.py - NO SQL HERE!

from database import init_db, get_user_by_id, create_user
def main(): init_db()
# Create user (no SQL!)
user_id = create_user("alice", "alice@example.com")
print(f"Created user: {user_id}")

# Get user (no SQL!)
user = get_user_by_id(user_id)
print(f"User: {user['username']}")
undefined
from database import init_db, get_user_by_id, create_user
def main(): init_db()
# Create user (no SQL!)
user_id = create_user("alice", "alice@example.com")
print(f"Created user: {user_id}")

# Get user (no SQL!)
user = get_user_by_id(user_id)
print(f"User: {user['username']}")
undefined

Anti-Patterns to Avoid

需避免的反模式

❌ WRONG: SQL in application code

❌ 错误示例：在应用代码中编写SQL

python
def process_user(user_id):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
python
def process_user(user_id):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))

✅ CORRECT: Use database functions

✅ 正确示例：使用数据库函数

python
def process_user(user_id):
    user = get_user_by_id(user_id)
python
def process_user(user_id):
    user = get_user_by_id(user_id)

Security: Always Use Parameters

安全提示：始终使用参数化查询

python
undefined
python
undefined

❌ WRONG: SQL injection vulnerability!

❌ WRONG: SQL injection vulnerability!

query = f"SELECT * FROM users WHERE email = '{email}'"
query = f"SELECT * FROM users WHERE email = '{email}'"

✅ CORRECT: Parameterized query

✅ CORRECT: Parameterized query

query = "SELECT * FROM users WHERE email = ?" results = db_query(query, (email,))
undefined
query = "SELECT * FROM users WHERE email = ?" results = db_query(query, (email,))
undefined

Common Patterns

常见模式

Pagination

分页查询

python
def get_users_paginated(page: int = 1, per_page: int = 10):
    offset = (page - 1) * per_page
    return db_query(
        "SELECT * FROM users ORDER BY created_at DESC LIMIT ? OFFSET ?",
        (per_page, offset)
    )
python
def get_users_paginated(page: int = 1, per_page: int = 10):
    offset = (page - 1) * per_page
    return db_query(
        "SELECT * FROM users ORDER BY created_at DESC LIMIT ? OFFSET ?",
        (per_page, offset)
    )

Transactions

事务处理

python
def db_transaction(operations: List[tuple]) -> bool:
    conn = get_connection()
    cursor = conn.cursor()
    try:
        for query, params in operations:
            cursor.execute(query, params)
        conn.commit()
        return True
    except:
        conn.rollback()
        return False
    finally:
        conn.close()
python
def db_transaction(operations: List[tuple]) -> bool:
    conn = get_connection()
    cursor = conn.cursor()
    try:
        for query, params in operations:
            cursor.execute(query, params)
        conn.commit()
        return True
    except:
        conn.rollback()
        return False
    finally:
        conn.close()

Golden Rules

黄金准则

  1. ✅ All SQL in database.py - nowhere else
  2. ✅ Parameterized queries - prevent SQL injection
  3. ✅ Meaningful return types - Optional, List, bool, int
  4. ✅ Transaction support - for multi-operation consistency
  5. ✅ *Add .db to .gitignore - don't commit databases
  1. ✅ 所有SQL都放在database.py中 - 禁止在其他文件编写
  2. ✅ 使用参数化查询 - 防止SQL注入
  3. ✅ 返回有意义的类型 - 比如Optional、List、bool、int
  4. ✅ 支持事务处理 - 保证多操作的一致性
  5. ✅ 将.db加入.gitignore* - 不要提交数据库文件到版本库

Related Resources

相关资源

See 
AgentUsage/db_usage.md
for complete documentation including:
  • Full database.py template
  • Migration patterns
  • Testing database functions
  • Performance optimization
请查看
AgentUsage/db_usage.md
获取完整文档，包括：
  • 完整的database.py模板
  • 数据库迁移模式
  • 数据库函数测试
  • 性能优化