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.

Core Architecture

Principles

• SQLite Only: Use SQLite as default database

• Single Interface: All database operations through database.py

• Complete SQL Isolation: All SQL statements in database.py

• Function-Based: Simple, reusable function interface

File Structure

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

Standard Interface Pattern

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

Domain-Specific Functions

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

Application Usage

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']}")

Anti-Patterns to Avoid

❌ WRONG: SQL in application code

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

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

Security: Always Use Parameters

❌ WRONG: SQL injection vulnerability!

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

✅ CORRECT: Parameterized query

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

Common Patterns

Pagination

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

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

• ✅ All SQL in database.py - nowhere else

• ✅ Parameterized queries - prevent SQL injection

• ✅ Meaningful return types - Optional, List, bool, int

• ✅ Transaction support - for multi-operation consistency

• ✅ *Add .db to .gitignore - don't commit databases

Related Resources

See AgentUsage/db_usage.md for complete documentation including:

• Full database.py template

• Migration patterns

• Testing database functions

• Performance optimization