Tutorial de Python SQLite3: Guia Completo de Banco de Dados SQLite em Python

import sqlite3
 
# Connect to a database file (creates it if it doesn't exist)
conn = sqlite3.connect("myapp.db")
 
# Create a cursor to execute SQL statements
cursor = conn.cursor()
 
# Always close the connection when done
conn.close()

import sqlite3
 
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()
 
cursor.execute("CREATE TABLE test (id INTEGER, value TEXT)")
cursor.execute("INSERT INTO test VALUES (1, 'hello')")
cursor.execute("SELECT * FROM test")
print(cursor.fetchone())  # (1, 'hello')
 
conn.close()  # Database is gone after this

import sqlite3
 
# Connection as context manager handles commit/rollback
with sqlite3.connect("myapp.db") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
    cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    # Auto-commits if no exception
    # Auto-rolls back if an exception occurs
 
# Note: the connection is NOT closed by 'with' -- it just commits/rolls back
# For full cleanup, close explicitly or use a wrapper:
conn.close()

import sqlite3
from contextlib import contextmanager
 
@contextmanager
def get_db(db_path="myapp.db"):
    """Context manager that commits on success, rolls back on error, and always closes."""
    conn = sqlite3.connect(db_path)
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()
 
# Usage
with get_db() as conn:
    conn.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
cursor.execute("""
    CREATE TABLE IF NOT EXISTS employees (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        department TEXT DEFAULT 'General',
        salary REAL CHECK(salary > 0),
        hire_date TEXT NOT NULL,
        is_active INTEGER DEFAULT 1
    )
""")
 
conn.commit()
conn.close()

import sqlite3
 
conn = sqlite3.connect("school.db")
cursor = conn.cursor()
 
# Enable foreign key enforcement (off by default in SQLite)
cursor.execute("PRAGMA foreign_keys = ON")
 
cursor.execute("""
    CREATE TABLE IF NOT EXISTS students (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE
    )
""")
 
cursor.execute("""
    CREATE TABLE IF NOT EXISTS courses (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        credits INTEGER DEFAULT 3
    )
""")
 
cursor.execute("""
    CREATE TABLE IF NOT EXISTS enrollments (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        student_id INTEGER NOT NULL,
        course_id INTEGER NOT NULL,
        grade TEXT,
        enrolled_date TEXT DEFAULT CURRENT_DATE,
        FOREIGN KEY (student_id) REFERENCES students(id) ON DELETE CASCADE,
        FOREIGN KEY (course_id) REFERENCES courses(id) ON DELETE CASCADE,
        UNIQUE(student_id, course_id)
    )
""")
 
conn.commit()
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Insert a single row
cursor.execute(
    "INSERT INTO employees (name, email, department, salary, hire_date) VALUES (?, ?, ?, ?, ?)",
    ("Alice Johnson", "alice@company.com", "Engineering", 95000, "2024-01-15")
)
 
# Insert multiple rows with executemany
employees = [
    ("Bob Smith", "bob@company.com", "Marketing", 72000, "2024-03-01"),
    ("Carol White", "carol@company.com", "Engineering", 98000, "2023-11-20"),
    ("David Brown", "david@company.com", "Sales", 68000, "2024-06-10"),
    ("Eve Davis", "eve@company.com", "Engineering", 105000, "2023-08-05"),
]
 
cursor.executemany(
    "INSERT INTO employees (name, email, department, salary, hire_date) VALUES (?, ?, ?, ?, ?)",
    employees
)
 
# Get the last inserted row ID
print(f"Last inserted ID: {cursor.lastrowid}")
 
# Get the number of rows affected
print(f"Rows inserted: {cursor.rowcount}")
 
conn.commit()
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Insert or update if email already exists
cursor.execute("""
    INSERT INTO employees (name, email, department, salary, hire_date)
    VALUES (?, ?, ?, ?, ?)
    ON CONFLICT(email) DO UPDATE SET
        name = excluded.name,
        salary = excluded.salary
""", ("Alice Johnson", "alice@company.com", "Engineering", 100000, "2024-01-15"))
 
conn.commit()
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Select all rows
cursor.execute("SELECT * FROM employees")
all_rows = cursor.fetchall()
for row in all_rows:
    print(row)  # Returns tuples by default
 
# Select specific columns with a condition
cursor.execute(
    "SELECT name, salary FROM employees WHERE department = ?",
    ("Engineering",)
)
engineers = cursor.fetchall()
for name, salary in engineers:
    print(f"{name}: ${salary:,.0f}")
 
# Select with ordering and limit
cursor.execute("""
    SELECT name, salary FROM employees
    ORDER BY salary DESC
    LIMIT 3
""")
top_earners = cursor.fetchall()
print("Top 3 earners:", top_earners)
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
conn.row_factory = sqlite3.Row  # Set row factory
 
cursor = conn.cursor()
cursor.execute("SELECT * FROM employees WHERE department = ?", ("Engineering",))
 
for row in cursor.fetchall():
    print(dict(row))  # Convert Row to dict
    print(row["name"], row["salary"])  # Access by column name
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Update a single field
cursor.execute(
    "UPDATE employees SET salary = ? WHERE email = ?",
    (110000, "alice@company.com")
)
print(f"Rows updated: {cursor.rowcount}")
 
# Update with a calculation
cursor.execute("""
    UPDATE employees
    SET salary = salary * 1.10
    WHERE department = 'Engineering' AND salary < 100000
""")
print(f"Engineers with raises: {cursor.rowcount}")
 
# Conditional update
cursor.execute("""
    UPDATE employees
    SET is_active = 0
    WHERE hire_date < '2023-01-01'
""")
 
conn.commit()
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Delete specific rows
cursor.execute("DELETE FROM employees WHERE is_active = 0")
print(f"Rows deleted: {cursor.rowcount}")
 
# Delete with a subquery
cursor.execute("""
    DELETE FROM enrollments
    WHERE student_id NOT IN (SELECT id FROM students)
""")
 
# Delete all rows (truncate equivalent)
cursor.execute("DELETE FROM employees")
 
conn.commit()
conn.close()

import sqlite3
 
# DANGEROUS: Never do this
user_input = "'; DROP TABLE employees; --"
query = f"SELECT * FROM employees WHERE name = '{user_input}'"
# This would execute: SELECT * FROM employees WHERE name = ''; DROP TABLE employees; --'

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Style 1: Question mark placeholders (positional)
cursor.execute(
    "SELECT * FROM employees WHERE department = ? AND salary > ?",
    ("Engineering", 90000)
)
 
# Style 2: Named placeholders
cursor.execute(
    "SELECT * FROM employees WHERE department = :dept AND salary > :min_salary",
    {"dept": "Engineering", "min_salary": 90000}
)
 
# Named placeholders are clearer with many parameters
cursor.execute("""
    INSERT INTO employees (name, email, department, salary, hire_date)
    VALUES (:name, :email, :dept, :salary, :date)
""", {
    "name": "Frank Green",
    "email": "frank@company.com",
    "dept": "Engineering",
    "salary": 92000,
    "date": "2025-09-01"
})
 
conn.commit()
conn.close()

import sqlite3
 
def search_employees(department=None, min_salary=None, is_active=None):
    conn = sqlite3.connect("myapp.db")
    cursor = conn.cursor()
 
    conditions = []
    params = []
 
    if department is not None:
        conditions.append("department = ?")
        params.append(department)
 
    if min_salary is not None:
        conditions.append("salary >= ?")
        params.append(min_salary)
 
    if is_active is not None:
        conditions.append("is_active = ?")
        params.append(is_active)
 
    query = "SELECT * FROM employees"
    if conditions:
        query += " WHERE " + " AND ".join(conditions)
 
    cursor.execute(query, params)
    results = cursor.fetchall()
    conn.close()
    return results
 
# Usage
engineers = search_employees(department="Engineering", min_salary=90000)
active_staff = search_employees(is_active=1)
everyone = search_employees()  # No filters

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
try:
    # Transfer money between accounts
    cursor.execute("UPDATE accounts SET balance = balance - 500 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 500 WHERE id = 2")
 
    # Verify no negative balance
    cursor.execute("SELECT balance FROM accounts WHERE id = 1")
    balance = cursor.fetchone()[0]
    if balance < 0:
        raise ValueError("Insufficient funds")
 
    conn.commit()  # All good, save changes
    print("Transfer complete")
 
except Exception as e:
    conn.rollback()  # Something went wrong, undo everything
    print(f"Transfer failed: {e}")
 
finally:
    conn.close()

import sqlite3
 
# Default behavior: transactions are deferred
conn = sqlite3.connect("myapp.db")
 
# Autocommit mode (each statement is its own transaction)
conn = sqlite3.connect("myapp.db", isolation_level=None)
 
# Immediate locking (prevents concurrent write conflicts)
conn = sqlite3.connect("myapp.db")
conn.execute("BEGIN IMMEDIATE")
# ... your operations ...
conn.commit()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
try:
    cursor.execute("INSERT INTO orders (customer_id, total) VALUES (1, 100)")
 
    # Start a savepoint
    cursor.execute("SAVEPOINT order_items")
 
    try:
        cursor.execute("INSERT INTO order_items (order_id, product, qty) VALUES (1, 'Widget', 5)")
        cursor.execute("INSERT INTO order_items (order_id, product, qty) VALUES (1, 'Gadget', -1)")
        # This might fail due to CHECK constraint
        cursor.execute("RELEASE SAVEPOINT order_items")
    except sqlite3.IntegrityError:
        cursor.execute("ROLLBACK TO SAVEPOINT order_items")
        print("Some items failed, but order was still created")
 
    conn.commit()
except Exception:
    conn.rollback()
finally:
    conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
cursor.execute("SELECT * FROM employees ORDER BY salary DESC")
 
# fetchone: Get the next single row (or None if no more rows)
top_earner = cursor.fetchone()
print(f"Top earner: {top_earner}")
 
# fetchmany: Get the next N rows
next_three = cursor.fetchmany(3)
print(f"Next 3: {next_three}")
 
# fetchall: Get all remaining rows
remaining = cursor.fetchall()
print(f"Remaining: {len(remaining)} rows")
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
cursor.execute("SELECT name, salary FROM employees")
 
# Memory-efficient: processes one row at a time
for name, salary in cursor:
    print(f"{name}: ${salary:,.0f}")
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("large_data.db")
cursor = conn.cursor()
 
cursor.execute("SELECT * FROM sensor_readings")
 
batch_size = 1000
while True:
    rows = cursor.fetchmany(batch_size)
    if not rows:
        break
    # Process batch
    for row in rows:
        pass  # your processing logic
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("school.db")
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
 
# INNER JOIN: Students with their enrolled courses
cursor.execute("""
    SELECT s.name AS student, c.title AS course, e.grade
    FROM enrollments e
    INNER JOIN students s ON e.student_id = s.id
    INNER JOIN courses c ON e.course_id = c.id
    ORDER BY s.name, c.title
""")
 
for row in cursor.fetchall():
    print(f"{row['student']} - {row['course']}: {row['grade'] or 'In Progress'}")
 
# LEFT JOIN: All students, even those not enrolled
cursor.execute("""
    SELECT s.name, COUNT(e.id) AS course_count
    FROM students s
    LEFT JOIN enrollments e ON s.id = e.student_id
    GROUP BY s.id
    ORDER BY course_count DESC
""")
 
for row in cursor.fetchall():
    print(f"{row['name']}: {row['course_count']} courses")
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Department statistics
cursor.execute("""
    SELECT
        department,
        COUNT(*) AS headcount,
        ROUND(AVG(salary), 2) AS avg_salary,
        MIN(salary) AS min_salary,
        MAX(salary) AS max_salary,
        SUM(salary) AS total_payroll
    FROM employees
    WHERE is_active = 1
    GROUP BY department
    HAVING COUNT(*) >= 2
    ORDER BY avg_salary DESC
""")
 
print(f"{'Department':<15} {'Count':<8} {'Avg Salary':<12} {'Min':<10} {'Max':<10}")
print("-" * 55)
for row in cursor.fetchall():
    dept, count, avg_sal, min_sal, max_sal, total = row
    print(f"{dept:<15} {count:<8} ${avg_sal:<11,.0f} ${min_sal:<9,.0f} ${max_sal:<9,.0f}")
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Employees earning above their department average
cursor.execute("""
    SELECT name, department, salary
    FROM employees e
    WHERE salary > (
        SELECT AVG(salary) FROM employees
        WHERE department = e.department
    )
    ORDER BY department, salary DESC
""")
 
for name, dept, salary in cursor.fetchall():
    print(f"{name} ({dept}): ${salary:,.0f}")
 
# Most recent hire in each department
cursor.execute("""
    SELECT name, department, hire_date
    FROM employees
    WHERE (department, hire_date) IN (
        SELECT department, MAX(hire_date)
        FROM employees
        GROUP BY department
    )
""")
 
for row in cursor.fetchall():
    print(row)
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()
 
# Rank employees by salary within each department
cursor.execute("""
    SELECT
        name,
        department,
        salary,
        RANK() OVER (PARTITION BY department ORDER BY salary DESC) AS dept_rank,
        RANK() OVER (ORDER BY salary DESC) AS overall_rank
    FROM employees
    WHERE is_active = 1
""")
 
for name, dept, salary, dept_rank, overall_rank in cursor.fetchall():
    print(f"#{overall_rank} overall, #{dept_rank} in {dept}: {name} (${salary:,.0f})")
 
# Running total of payroll by hire date
cursor.execute("""
    SELECT
        name,
        hire_date,
        salary,
        SUM(salary) OVER (ORDER BY hire_date) AS running_total
    FROM employees
    ORDER BY hire_date
""")
 
for name, date, salary, running in cursor.fetchall():
    print(f"{date} | {name}: ${salary:,.0f} | Running total: ${running:,.0f}")
 
conn.close()

import sqlite3
import pandas as pd
 
conn = sqlite3.connect("myapp.db")
 
# Read a SQL query into a DataFrame
df = pd.read_sql("SELECT * FROM employees WHERE is_active = 1", conn)
print(df.head())
print(df.describe())
 
# Parameterized query with pandas
df = pd.read_sql(
    "SELECT * FROM employees WHERE department = ? AND salary > ?",
    conn,
    params=("Engineering", 90000)
)
 
# Read an entire table
df = pd.read_sql("SELECT * FROM employees", conn, index_col="id")
 
conn.close()

import sqlite3
import pandas as pd
 
# Create a sample DataFrame
df = pd.DataFrame({
    "product": ["Laptop", "Phone", "Tablet", "Monitor", "Keyboard"],
    "price": [999.99, 699.99, 449.99, 349.99, 79.99],
    "stock": [50, 200, 75, 30, 500],
    "category": ["Electronics", "Electronics", "Electronics", "Peripherals", "Peripherals"]
})
 
conn = sqlite3.connect("store.db")
 
# Write DataFrame to a new table
df.to_sql("products", conn, if_exists="replace", index=False)
 
# Append to an existing table
new_products = pd.DataFrame({
    "product": ["Mouse", "Webcam"],
    "price": [49.99, 89.99],
    "stock": [300, 100],
    "category": ["Peripherals", "Peripherals"]
})
new_products.to_sql("products", conn, if_exists="append", index=False)
 
# Verify
result = pd.read_sql("SELECT * FROM products", conn)
print(result)
 
conn.close()

import sqlite3
import pandas as pd
 
# Step 1: Load CSV into SQLite
df = pd.read_csv("sales_data.csv")
conn = sqlite3.connect("analytics.db")
df.to_sql("sales", conn, if_exists="replace", index=False)
 
# Step 2: Run SQL analytics
summary = pd.read_sql("""
    SELECT
        region,
        strftime('%Y-%m', sale_date) AS month,
        COUNT(*) AS transactions,
        SUM(amount) AS revenue,
        AVG(amount) AS avg_transaction
    FROM sales
    GROUP BY region, month
    ORDER BY month, revenue DESC
""", conn)
 
print(summary)
 
# Step 3: Export results
summary.to_csv("monthly_summary.csv", index=False)
 
conn.close()

import pygwalker as pyg
 
# Turn your SQL query results into an interactive visualization
walker = pyg.walk(summary)

import sqlite3
 
def safe_insert(db_path, name, email, salary):
    """Insert an employee with proper error handling."""
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO employees (name, email, salary, hire_date) VALUES (?, ?, ?, date('now'))",
            (name, email, salary)
        )
        conn.commit()
        return cursor.lastrowid
 
    except sqlite3.IntegrityError as e:
        # Unique constraint violated (duplicate email, NOT NULL violated, etc.)
        print(f"Data integrity error: {e}")
        return None
 
    except sqlite3.OperationalError as e:
        # Table doesn't exist, SQL syntax error, database locked, etc.
        print(f"Operational error: {e}")
        return None
 
    except sqlite3.DatabaseError as e:
        # Corrupted database, disk full, etc.
        print(f"Database error: {e}")
        return None
 
    except sqlite3.ProgrammingError as e:
        # Using a closed cursor, wrong number of parameters, etc.
        print(f"Programming error: {e}")
        return None
 
    finally:
        if conn:
            conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
conn.execute("PRAGMA journal_mode=WAL")
# Returns 'wal' if successful

import sqlite3
import time
 
conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE numbers (value INTEGER)")
 
data = [(i,) for i in range(100_000)]
 
# Slow: individual inserts
start = time.perf_counter()
for d in data:
    conn.execute("INSERT INTO numbers VALUES (?)", d)
conn.commit()
slow_time = time.perf_counter() - start
 
conn.execute("DELETE FROM numbers")
 
# Fast: executemany
start = time.perf_counter()
conn.executemany("INSERT INTO numbers VALUES (?)", data)
conn.commit()
fast_time = time.perf_counter() - start
 
print(f"Individual inserts: {slow_time:.2f}s")
print(f"executemany:        {fast_time:.2f}s")
print(f"Speedup:            {slow_time / fast_time:.1f}x")
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
 
# Single column index
conn.execute("CREATE INDEX IF NOT EXISTS idx_employees_department ON employees(department)")
 
# Composite index (for queries that filter on both columns)
conn.execute("CREATE INDEX IF NOT EXISTS idx_employees_dept_salary ON employees(department, salary)")
 
# Unique index (also enforces uniqueness)
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_employees_email ON employees(email)")
 
# Check existing indexes
cursor = conn.execute("SELECT name, sql FROM sqlite_master WHERE type='index'")
for name, sql in cursor.fetchall():
    print(f"{name}: {sql}")
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
 
# Check how SQLite executes a query
result = conn.execute("""
    EXPLAIN QUERY PLAN
    SELECT * FROM employees WHERE department = 'Engineering' AND salary > 90000
""").fetchall()
 
for row in result:
    print(row)
# Output shows whether an index is used or a full table scan occurs
 
conn.close()

import sqlite3
 
conn = sqlite3.connect("myapp.db")
 
# Increase cache size (default is 2000 pages = ~8MB)
conn.execute("PRAGMA cache_size = -64000")  # 64MB (negative = KB)
 
# Reduce disk syncs for speed (less durable but much faster)
conn.execute("PRAGMA synchronous = NORMAL")  # Options: OFF, NORMAL, FULL (default)
 
# Store temp tables in memory
conn.execute("PRAGMA temp_store = MEMORY")
 
# Check database integrity
result = conn.execute("PRAGMA integrity_check").fetchone()
print(f"Integrity: {result[0]}")  # Should print 'ok'
 
conn.close()

import sqlite3
import sys
from datetime import datetime, date
from contextlib import contextmanager
 
DB_PATH = "tasks.db"
 
@contextmanager
def get_db():
    """Get a database connection with Row factory and foreign keys enabled."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA journal_mode = WAL")
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()
 
def init_db():
    """Create tables if they don't exist."""
    with get_db() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                description TEXT DEFAULT '',
                priority INTEGER DEFAULT 2 CHECK(priority BETWEEN 1 AND 4),
                status TEXT DEFAULT 'todo' CHECK(status IN ('todo', 'in_progress', 'done')),
                due_date TEXT,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP,
                completed_at TEXT
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS tags (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT UNIQUE NOT NULL
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS task_tags (
                task_id INTEGER NOT NULL,
                tag_id INTEGER NOT NULL,
                PRIMARY KEY (task_id, tag_id),
                FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE,
                FOREIGN KEY (tag_id) REFERENCES tags(id) ON DELETE CASCADE
            )
        """)
        conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_priority ON tasks(priority)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_due ON tasks(due_date)")
 
def add_task(title, description="", priority=2, due_date=None, tags=None):
    """Add a new task with optional tags."""
    with get_db() as conn:
        cursor = conn.execute(
            "INSERT INTO tasks (title, description, priority, due_date) VALUES (?, ?, ?, ?)",
            (title, description, priority, due_date)
        )
        task_id = cursor.lastrowid
 
        if tags:
            for tag_name in tags:
                conn.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (tag_name,))
                tag_id = conn.execute("SELECT id FROM tags WHERE name = ?", (tag_name,)).fetchone()["id"]
                conn.execute("INSERT INTO task_tags (task_id, tag_id) VALUES (?, ?)", (task_id, tag_id))
 
        print(f"Task #{task_id} created: {title}")
        return task_id
 
def list_tasks(status=None, priority=None, tag=None):
    """List tasks with optional filters."""
    with get_db() as conn:
        conditions = []
        params = []
 
        if status:
            conditions.append("t.status = ?")
            params.append(status)
        if priority:
            conditions.append("t.priority = ?")
            params.append(priority)
        if tag:
            conditions.append("t.id IN (SELECT task_id FROM task_tags tt JOIN tags tg ON tt.tag_id = tg.id WHERE tg.name = ?)")
            params.append(tag)
 
        query = """
            SELECT t.id, t.title, t.priority, t.status, t.due_date,
                   GROUP_CONCAT(tg.name, ', ') AS tags
            FROM tasks t
            LEFT JOIN task_tags tt ON t.id = tt.task_id
            LEFT JOIN tags tg ON tt.tag_id = tg.id
        """
        if conditions:
            query += " WHERE " + " AND ".join(conditions)
        query += " GROUP BY t.id ORDER BY t.priority ASC, t.due_date ASC"
 
        rows = conn.execute(query, params).fetchall()
 
        priority_labels = {1: "URGENT", 2: "HIGH", 3: "MEDIUM", 4: "LOW"}
        print(f"\n{'ID':<5} {'Priority':<10} {'Status':<13} {'Due':<12} {'Title':<30} {'Tags'}")
        print("-" * 85)
        for row in rows:
            due = row["due_date"] or "—"
            tags_str = row["tags"] or "—"
            p_label = priority_labels.get(row["priority"], "?")
            print(f"{row['id']:<5} {p_label:<10} {row['status']:<13} {due:<12} {row['title']:<30} {tags_str}")
 
        print(f"\nTotal: {len(rows)} tasks")
 
def complete_task(task_id):
    """Mark a task as done."""
    with get_db() as conn:
        result = conn.execute(
            "UPDATE tasks SET status = 'done', completed_at = ? WHERE id = ?",
            (datetime.now().isoformat(), task_id)
        )
        if result.rowcount:
            print(f"Task #{task_id} marked as done.")
        else:
            print(f"Task #{task_id} not found.")
 
def stats():
    """Show task statistics."""
    with get_db() as conn:
        row = conn.execute("""
            SELECT
                COUNT(*) AS total,
                SUM(CASE WHEN status = 'todo' THEN 1 ELSE 0 END) AS todo,
                SUM(CASE WHEN status = 'in_progress' THEN 1 ELSE 0 END) AS in_progress,
                SUM(CASE WHEN status = 'done' THEN 1 ELSE 0 END) AS done,
                SUM(CASE WHEN due_date < date('now') AND status != 'done' THEN 1 ELSE 0 END) AS overdue
            FROM tasks
        """).fetchone()
 
        print(f"\nTask Statistics:")
        print(f"  Total:       {row['total']}")
        print(f"  To Do:       {row['todo']}")
        print(f"  In Progress: {row['in_progress']}")
        print(f"  Done:        {row['done']}")
        print(f"  Overdue:     {row['overdue']}")
 
# Initialize and demo
if __name__ == "__main__":
    init_db()
 
    # Add sample tasks
    add_task("Design database schema", priority=1, due_date="2026-02-20", tags=["backend", "database"])
    add_task("Write API endpoints", priority=2, due_date="2026-02-25", tags=["backend", "api"])
    add_task("Create unit tests", priority=3, tags=["testing"])
    add_task("Update documentation", priority=4, due_date="2026-03-01", tags=["docs"])
 
    # List and stats
    list_tasks()
    stats()
    complete_task(1)
    list_tasks(status="done")

import sqlite3
 
source = sqlite3.connect("myapp.db")
backup = sqlite3.connect("myapp_backup.db")
source.backup(backup)
backup.close()
source.close()