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