SQLite Database Manager (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: SQLite Database Manager

Write a program that provides a command-line interface for managing SQLite databases, supporting table creation, CRUD operations, schema inspection, data import/export, and query execution.

FUNCTIONAL REQUIREMENTS:
- Accept a database file path as a command-line argument (create it if it does not exist)
- Support creating tables with typed columns (TEXT, INTEGER, REAL, BLOB) and constraints (PRIMARY KEY, NOT NULL, UNIQUE, DEFAULT)
- Support INSERT, SELECT, UPDATE, and DELETE operations via command-line subcommands
- Support executing arbitrary SQL queries via a --query flag
- Implement schema inspection: list all tables, show column details for a given table, show row counts
- Import data from a CSV file into a specified table via --import flag, auto-detecting column types
- Export a table or query result to CSV via --export flag
- Support basic transactions: --begin, --commit, --rollback flags for grouping operations
- Print query results as formatted tables to console with column headers and aligned values
- If no database path is given, create a sample database with at least three related tables (e.g., users, orders, products), populate with sample data, and run demonstration queries showing joins, aggregations, and filtering
- Handle database errors gracefully with clear messages distinguishing connection, schema, constraint, and query errors

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# SQLite Database Manager (Python)

Command-line SQLite manager for Python 3.10+ using the built-in `sqlite3` module.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies

- Direct: none
- Transitive: none

`requirements.txt` is intentionally empty of packages because this implementation uses only the Python standard library.

## Run

No DB path: creates a sample database and runs demo queries.

```bash
python src/main.py
```

Use a specific DB:

```bash
python src/main.py ./app.db --list-tables
```

## Commands

Create table:

```bash
python src/main.py ./app.db --create-table users --columns "id:INTEGER PRIMARY KEY,name:TEXT NOT NULL,email:TEXT UNIQUE"
```

Insert:

```bash
python src/main.py ./app.db --insert users --values "id=1,name=Alice,email=alice@example.com"
```

Select:

```bash
python src/main.py ./app.db --select users --columns "id,name,email" --where "id >= 1" --limit 10
```

Update:

```bash
python src/main.py ./app.db --update users --set "name=Alice Updated" --where "id = 1"
```

Delete:

```bash
python src/main.py ./app.db --delete users --where "id = 1"
```

Arbitrary query:

```bash
python src/main.py ./app.db --query "SELECT name FROM sqlite_master WHERE type='table'"
```

Schema inspection:

```bash
python src/main.py ./app.db --list-tables
python src/main.py ./app.db --describe users
python src/main.py ./app.db --row-count users
```

Import CSV:

```bash
python src/main.py ./app.db --import ./users.csv --table imported_users
```

Export to CSV:

```bash
python src/main.py ./app.db --export ./users_out.csv --table users
python src/main.py ./app.db --export ./query_out.csv --query "SELECT * FROM users WHERE id > 5"
```

Transactions:

```bash
python src/main.py ./app.db --begin --insert users --values "id=3,name=Carol,email=carol@example.com" --commit
python src/main.py ./app.db --begin --delete users --where "id=3" --rollback
```

## Expected Output

- Aligned table output for query results
- Automatic DB creation if missing
- Clear categorized error messages (`Connection`, `Schema`, `Constraint`, `Query`)

# No external dependencies required.
# Direct dependencies: none
# Transitive dependencies: none

#!/usr/bin/env python3
import argparse
import csv
import os
import sqlite3
import sys
from pathlib import Path
from typing import Any, Iterable


def classify_error(message: str) -> str:
    lower = message.lower()
    if "no such table" in lower or "no such column" in lower or "already exists" in lower:
        return "Schema Error"
    if "constraint failed" in lower or "not null" in lower or "unique" in lower:
        return "Constraint Error"
    if "syntax error" in lower or "near" in lower:
        return "Query Error"
    return "Connection Error"


def escape_ident(name: str) -> str:
    return '"' + name.replace('"', '""') + '"'


def parse_key_value_spec(spec: str) -> list[tuple[str, Any]]:
    parts = next(csv.reader([spec], skipinitialspace=True))
    output: list[tuple[str, Any]] = []
    for part in parts:
        if "=" not in part:
            raise ValueError(f"Invalid key=value pair: {part}")
        key, value = part.split("=", 1)
        output.append((key.strip(), parse_value(value.strip())))
    return output


def parse_column_spec(spec: str) -> list[tuple[str, str]]:
    parts = next(csv.reader([spec], skipinitialspace=True))
    output: list[tuple[str, str]] = []
    for part in parts:
        if ":" not in part:
            raise ValueError(f"Invalid column definition: {part}")
        name, definition = part.split(":", 1)
        name = name.strip()
        definition = definition.strip()
        if not name or not definition:
            raise ValueError(f"Invalid column definition: {part}")
        output.append((name, definition))
    return output


def parse_value(raw: str | None) -> Any:
    if raw is None:
        return None
    text = raw.strip()
    if text == "":
        return ""
    if text.lower() == "null":
        return None
    if text.lower() == "true":
        return 1
    if text.lower() == "false":
        return 0
    try:
        if "." in text:
            return float(text)
        return int(text)
    except ValueError:
        pass
    if (text.startswith('"') and text.endswith('"')) or (text.startswith("'") and text.endswith("'")):
        return text[1:-1]
    return text


def print_table(columns: list[str], rows: Iterable[Iterable[Any]]) -> None:
    rows_list = [list(row) for row in rows]
    if not columns:
        print("(no columns)")
        return
    widths = [len(col) for col in columns]
    for row in rows_list:
        for i, value in enumerate(row):
            text = "NULL" if value is None else str(value)
            if len(text) > widths[i]:
                widths[i] = len(text)

    def pad(text: str, width: int) -> str:
        return text + (" " * max(0, width - len(text)))

    sep = "+-" + "-+-".join("-" * width for width in widths) + "-+"
    print(sep)
    print("| " + " | ".join(pad(columns[i], widths[i]) for i in range(len(columns))) + " |")
    print(sep)
    for row in rows_list:
        print("| " + " | ".join(pad("NULL" if v is None else str(v), widths[i]) for i, v in enumerate(row)) + " |")
    print(sep)
    print(f"{len(rows_list)} row(s)")


def detect_type(values: list[str]) -> str:
    non_empty = [v.strip() for v in values if v is not None and v.strip() != ""]
    if not non_empty:
        return "TEXT"
    if all(v.lstrip("-").isdigit() for v in non_empty):
        return "INTEGER"
    if all(_is_float(v) for v in non_empty):
        return "REAL"
    if all(v.lower().startswith("0x") for v in non_empty):
        return "BLOB"
    return "TEXT"


def _is_float(value: str) -> bool:
    try:
        float(value)
        return True
    except ValueError:
        return False


def table_exists(conn: sqlite3.Connection, table: str) -> bool:
    cur = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name = ?", (table,))
    return cur.fetchone() is not None


def run_query_and_print(conn: sqlite3.Connection, sql: str) -> tuple[list[str], list[tuple[Any, ...]]]:
    cur = conn.execute(sql)
    if cur.description is None:
        print("Query executed.")
        return [], []
    columns = [c[0] for c in cur.description]
    rows = cur.fetchall()
    print_table(columns, rows)
    return columns, rows


def inspect_schema(conn: sqlite3.Connection, args: argparse.Namespace) -> None:
    if args.list_tables:
        cur = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")
        names = [row[0] for row in cur.fetchall()]
        if not names:
            print("No user tables found.")
        else:
            rows = []
            for name in names:
                count = conn.execute(f"SELECT COUNT(*) FROM {escape_ident(name)}").fetchone()[0]
                rows.append((name, count))
            print_table(["table", "row_count"], rows)

    if args.describe:
        cur = conn.execute(f"PRAGMA table_info({escape_ident(args.describe)})")
        rows = cur.fetchall()
        columns = [c[0] for c in cur.description] if cur.description else []
        print_table(columns, rows)

    if args.row_count:
        cur = conn.execute(f"SELECT COUNT(*) AS row_count FROM {escape_ident(args.row_count)}")
        rows = cur.fetchall()
        columns = [c[0] for c in cur.description] if cur.description else []
        print_table(columns, rows)


def do_create_table(conn: sqlite3.Connection, args: argparse.Namespace) -> bool:
    if not args.create_table:
        return False
    if not args.columns:
        raise ValueError("--create-table requires --columns")
    defs = parse_column_spec(args.columns)
    sql = f"CREATE TABLE IF NOT EXISTS {escape_ident(args.create_table)} (" + ", ".join(
        f"{escape_ident(name)} {definition}" for name, definition in defs
    ) + ")"
    conn.execute(sql)
    print(f"Created/verified table: {args.create_table}")
    return True


def do_insert(conn: sqlite3.Connection, args: argparse.Namespace) -> bool:
    if not args.insert:
        return False
    if not args.values:
        raise ValueError("--insert requires --values")
    entries = parse_key_value_spec(args.values)
    columns = [key for key, _ in entries]
    values = [value for _, value in entries]
    sql = (
        f"INSERT INTO {escape_ident(args.insert)} (" + ", ".join(escape_ident(c) for c in columns) + ") VALUES ("
        + ", ".join("?" for _ in values)
        + ")"
    )
    conn.execute(sql, values)
    print(f"Inserted 1 row into {args.insert}")
    return True


def do_select(conn: sqlite3.Connection, args: argparse.Namespace) -> bool:
    if not args.select:
        return False
    columns = args.columns.split(",") if args.columns else ["*"]
    columns = [c.strip() for c in columns if c.strip()]
    if not columns:
        columns = ["*"]
    sql = f"SELECT {', '.join(columns)} FROM {escape_ident(args.select)}"
    if args.where:
        sql += f" WHERE {args.where}"
    if args.limit:
        sql += f" LIMIT {int(args.limit)}"
    run_query_and_print(conn, sql)
    return True


def do_update(conn: sqlite3.Connection, args: argparse.Namespace) -> bool:
    if not args.update:
        return False
    if not args.set:
        raise ValueError("--update requires --set")
    entries = parse_key_value_spec(args.set)
    sql = f"UPDATE {escape_ident(args.update)} SET " + ", ".join(f"{escape_ident(k)} = ?" for k, _ in entries)
    if args.where:
        sql += f" WHERE {args.where}"
    conn.execute(sql, [v for _, v in entries])
    changed = conn.execute("SELECT changes()").fetchone()[0]
    print(f"Updated {changed} row(s) in {args.update}")
    return True


def do_delete(conn: sqlite3.Connection, args: argparse.Namespace) -> bool:
    if not args.delete:
        return False
    sql = f"DELETE FROM {escape_ident(args.delete)}"
    if args.where:
        sql += f" WHERE {args.where}"
    conn.execute(sql)
    changed = conn.execute("SELECT changes()").fetchone()[0]
    print(f"Deleted {changed} row(s) from {args.delete}")
    return True


def do_import(conn: sqlite3.Connection, args: argparse.Namespace) -> bool:
    if not args.import_file:
        return False
    if not args.table:
        raise ValueError("--import requires --table")
    csv_path = Path(args.import_file).resolve()
    if not csv_path.exists():
        raise FileNotFoundError(f"CSV file not found: {csv_path}")

    with csv_path.open("r", newline="", encoding="utf-8") as f:
        reader = list(csv.reader(f))
    if not reader:
        raise ValueError("CSV is empty.")
    headers = [h.strip() for h in reader[0]]
    data_rows = reader[1:]

    if not table_exists(conn, args.table):
        types = []
        for idx, header in enumerate(headers):
            values = [row[idx] if idx < len(row) else "" for row in data_rows]
            types.append((header, detect_type(values)))
        create_sql = f"CREATE TABLE {escape_ident(args.table)} (" + ", ".join(
            f"{escape_ident(name)} {column_type}" for name, column_type in types
        ) + ")"
        conn.execute(create_sql)

    if data_rows:
        placeholders = ", ".join("?" for _ in headers)
        insert_sql = f"INSERT INTO {escape_ident(args.table)} (" + ", ".join(escape_ident(h) for h in headers) + f") VALUES ({placeholders})"
        formatted_rows = [
            [parse_value(row[idx]) if idx < len(row) else None for idx in range(len(headers))]
            for row in data_rows
        ]
        conn.executemany(insert_sql, formatted_rows)

    print(f"Imported {len(data_rows)} row(s) into {args.table}")
    return True


def do_export(conn: sqlite3.Connection, args: argparse.Namespace) -> bool:
    if not args.export:
        return False
    output = Path(args.export).resolve()
    if args.query:
        cur = conn.execute(args.query)
    elif args.table:
        cur = conn.execute(f"SELECT * FROM {escape_ident(args.table)}")
    else:
        raise ValueError("--export requires either --table or --query")

    if cur.description is None:
        raise ValueError("Export query did not produce rows.")

    columns = [c[0] for c in cur.description]
    rows = cur.fetchall()
    with output.open("w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(columns)
        writer.writerows(rows)
    print(f"Exported {len(rows)} row(s) to {output}")
    return True


def build_sample_db(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        PRAGMA foreign_keys = ON;
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL
        );
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            price REAL NOT NULL CHECK(price >= 0)
        );
        CREATE TABLE IF NOT EXISTS orders (
            id INTEGER PRIMARY KEY,
            user_id INTEGER NOT NULL REFERENCES users(id),
            product_id INTEGER NOT NULL REFERENCES products(id),
            quantity INTEGER NOT NULL DEFAULT 1,
            created_at TEXT NOT NULL
        );
        DELETE FROM orders;
        DELETE FROM users;
        DELETE FROM products;
        """
    )

    conn.executemany(
        "INSERT INTO users(id, name, email) VALUES (?, ?, ?)",
        [(1, "Alice", "alice@example.com"), (2, "Bob", "bob@example.com"), (3, "Carol", "carol@example.com")],
    )
    conn.executemany(
        "INSERT INTO products(id, name, price) VALUES (?, ?, ?)",
        [(1, "Keyboard", 79.99), (2, "Monitor", 249.50), (3, "Mouse", 29.95)],
    )
    conn.executemany(
        "INSERT INTO orders(id, user_id, product_id, quantity, created_at) VALUES (?, ?, ?, ?, ?)",
        [
            (1, 1, 1, 2, "2026-02-01T10:00:00Z"),
            (2, 1, 2, 1, "2026-02-02T11:30:00Z"),
            (3, 2, 3, 3, "2026-02-03T15:45:00Z"),
            (4, 3, 2, 1, "2026-02-04T17:15:00Z"),
        ],
    )


def run_demo(conn: sqlite3.Connection) -> None:
    print("No database path provided. Creating sample database and running demo queries.\n")
    build_sample_db(conn)
    print("Demo 1: Join users, orders, products")
    run_query_and_print(
        conn,
        """
        SELECT o.id AS order_id, u.name AS user_name, p.name AS product_name, o.quantity, p.price,
               ROUND(o.quantity * p.price, 2) AS total
        FROM orders o
        JOIN users u ON u.id = o.user_id
        JOIN products p ON p.id = o.product_id
        ORDER BY o.id
        """,
    )
    print("\nDemo 2: Aggregation by user")
    run_query_and_print(
        conn,
        """
        SELECT u.name, ROUND(SUM(o.quantity * p.price), 2) AS spent
        FROM users u
        JOIN orders o ON o.user_id = u.id
        JOIN products p ON p.id = o.product_id
        GROUP BY u.id, u.name
        ORDER BY spent DESC
        """,
    )
    print("\nDemo 3: Filtering products over 50")
    run_query_and_print(conn, "SELECT id, name, price FROM products WHERE price > 50 ORDER BY price DESC")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="SQLite Database Manager")
    parser.add_argument("db_path", nargs="?", help="Database file path")

    parser.add_argument("--create-table", dest="create_table")
    parser.add_argument("--columns")

    parser.add_argument("--insert")
    parser.add_argument("--values")

    parser.add_argument("--select")
    parser.add_argument("--where")
    parser.add_argument("--limit")

    parser.add_argument("--update")
    parser.add_argument("--set")
    parser.add_argument("--delete")
    parser.add_argument("--query")

    parser.add_argument("--list-tables", action="store_true")
    parser.add_argument("--describe")
    parser.add_argument("--row-count")

    parser.add_argument("--import", dest="import_file")
    parser.add_argument("--table")
    parser.add_argument("--export")

    parser.add_argument("--begin", action="store_true")
    parser.add_argument("--commit", action="store_true")
    parser.add_argument("--rollback", action="store_true")
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    db_path = Path(args.db_path).resolve() if args.db_path else Path("sample_demo.sqlite").resolve()
    os.makedirs(db_path.parent, exist_ok=True)

    try:
        conn = sqlite3.connect(db_path)
        conn.execute("PRAGMA foreign_keys = ON")
    except sqlite3.Error as exc:
        msg = str(exc)
        print(f"[{classify_error(msg)}] {msg}", file=sys.stderr)
        return 1

    mutated = not db_path.exists()
    operations = 0
    in_tx = False
    try:
        if not args.db_path:
            run_demo(conn)
            mutated = True
            operations += 1
        else:
            if args.begin:
                conn.execute("BEGIN")
                in_tx = True
                print("Transaction started.")

            if do_create_table(conn, args):
                mutated = True
                operations += 1
            if do_insert(conn, args):
                mutated = True
                operations += 1
            if do_update(conn, args):
                mutated = True
                operations += 1
            if do_delete(conn, args):
                mutated = True
                operations += 1
            if do_import(conn, args):
                mutated = True
                operations += 1
            if args.query:
                run_query_and_print(conn, args.query)
                operations += 1
            if do_select(conn, args):
                operations += 1
            if args.list_tables or args.describe or args.row_count:
                inspect_schema(conn, args)
                operations += 1
            if do_export(conn, args):
                operations += 1

            if args.rollback and in_tx:
                conn.rollback()
                print("Transaction rolled back.")
                mutated = False
                in_tx = False
            elif args.commit and in_tx:
                conn.commit()
                print("Transaction committed.")
                mutated = True
                in_tx = False
            elif in_tx:
                conn.commit()
                print("Transaction committed automatically.")
                mutated = True
                in_tx = False
            elif mutated:
                conn.commit()

        if operations == 0:
            parser.print_help()

        print(f"Database path: {db_path}")
        return 0
    except (sqlite3.Error, ValueError, FileNotFoundError) as exc:
        if in_tx:
            conn.rollback()
        msg = str(exc)
        print(f"[{classify_error(msg)}] {msg}", file=sys.stderr)
        return 1
    finally:
        conn.close()


if __name__ == "__main__":
    raise SystemExit(main())