# Python Advanced SQL

> Write Python decorators/context managers/dataclasses and query gene/variant tables with sqlite3/pandas SQL (JOIN, GROUP BY, HAVING). Use for retry/caching/validation wrappers or SQL against Ensembl/UCSC-style schemas.

- Skill: `pavel-kravchenko/python-advanced-sql` (Agent Skill)
- Install (CLI): `npx skillmds@latest add pavel-kravchenko/python-advanced-sql`
- Raw SKILL.md: https://api.skillmd.com/api/skills/pavel-kravchenko/python-advanced-sql/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Data & Analytics
- Author: pavel-kravchenko (https://skillmd.com/u/pavel-kravchenko)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/pavel-kravchenko/python-advanced-sql

---


# Advanced Python & SQL for Bioinformatics

## When to Use
- Modeling sequences, genes, variants, or alignments as Python classes (dunders, `@dataclass`)
- Adding caching, timing, input validation, or retry logic to pipeline functions via decorators
- Safely handling files, DB connections, or temp files with context managers (`with` / `contextlib`)
- Querying biological databases (SQLite, Ensembl MySQL dumps, UCSC tables) with SQL joins/aggregation
- Joining gene/variant/expression tables to answer "which genes are X and Y" questions

## Version Compatibility
- Python >= 3.10 (uses `dataclasses`, `functools`, PEP 604 type hints)
- `sqlite3` — stdlib, ships with Python (no install needed)
- `pandas` >= 2.0 (for `pd.read_sql_query`)

## Prerequisites
- `pip install pandas`
- Comfortable with plain functions and basic classes
- For fetching real Ensembl/UCSC/NCBI data before loading it into SQLite, see `bio-database-access-entrez-fetch` or `bio-database-access-batch-downloads`

## Quick Reference

### OOP Dunders
| Method | Purpose |
|--------|---------|
| `__init__` | Constructor |
| `__str__` / `__repr__` | User / debug string |
| `__len__` | `len(obj)` |
| `__eq__`, `__lt__` | Comparison / sorting |
| `__contains__` | `"ATG" in seq` syntax |
| `__enter__` / `__exit__` | Context manager |

### SQL Clauses
| Clause | Use |
|--------|-----|
| `WHERE biotype = 'protein_coding'` | Filter rows |
| `GROUP BY tissue, condition` | Aggregate groups |
| `HAVING AVG(tpm) > 50` | Filter after grouping |
| `INNER JOIN` | Matching rows only |
| `LEFT JOIN` | All left rows, NULLs for no match |
| Subquery with `IN (SELECT ...)` | Multi-condition filter |

## Key Patterns

### Decorators for pipeline functions

**Goal:** add cross-cutting behavior (timing, memoization, input validation, retry-on-failure) to pipeline functions without rewriting each one.

**Approach:** write a decorator factory that wraps the target function, always use `@functools.wraps` to preserve `__name__`/`__doc__`, and stack decorators bottom-up (the one closest to `def` runs first).

```python
import functools
import time


def timer(func):
    """Print the wall-clock time a function call took."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        t0 = time.perf_counter()
        result = func(*args, **kwargs)
        print(f"[timer] {func.__name__}: {time.perf_counter() - t0:.4f}s")
        return result
    return wrapper


def memoize(func):
    """Cache results by argument tuple; only safe for hashable args."""
    cache = {}
    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    wrapper.cache = cache
    return wrapper


def validate_sequence(valid_chars: str, seq_type: str = "DNA"):
    """Decorator factory: reject sequences with characters outside valid_chars."""
    valid_set = set(valid_chars.upper())
    def decorator(func):
        @functools.wraps(func)
        def wrapper(seq, *args, **kwargs):
            invalid = set(seq.upper()) - valid_set
            if invalid:
                raise ValueError(f"Invalid {seq_type} characters {invalid} in {func.__name__}()")
            return func(seq, *args, **kwargs)
        return wrapper
    return decorator


def retry(max_attempts: int = 3, delay: float = 0.5):
    """Decorator factory: retry a flaky call (e.g. a network fetch) with backoff."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_err = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_err = e
                    if attempt < max_attempts:
                        time.sleep(delay)
            raise last_err
        return wrapper
    return decorator


# Stacking: applied bottom-up -- validate_sequence runs first, timer wraps around it
@timer
@validate_sequence('ATGC', seq_type='DNA')
def gc_content(seq: str) -> float:
    """Return GC percentage of a validated DNA sequence."""
    seq = seq.upper()
    return (seq.count('G') + seq.count('C')) / len(seq) * 100
```

### Context managers for safe resource handling

**Goal:** guarantee files, DB connections, and temp files are closed/removed even when an exception is raised mid-pipeline.

**Approach:** implement `__enter__`/`__exit__` for stateful resources, or use `@contextlib.contextmanager` for simple one-shot setup/teardown.

```python
import os
import tempfile
from contextlib import contextmanager


class FastaWriter:
    """Class-based context manager: opens a FASTA file, wraps sequences at line_width."""

    def __init__(self, filename: str, line_width: int = 80):
        self.filename, self.line_width = filename, line_width
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, 'w')
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.file.close()
        return False  # never suppress exceptions

    def write_record(self, seq_id: str, seq: str, desc: str = ""):
        header = f">{seq_id}" + (f" {desc}" if desc else "")
        self.file.write(header + "\n")
        for i in range(0, len(seq), self.line_width):
            self.file.write(seq[i:i + self.line_width] + "\n")


@contextmanager
def temp_fasta(sequences: dict[str, str]):
    """Function-based context manager: write sequences to a temp FASTA, delete on exit."""
    fd, path = tempfile.mkstemp(suffix='.fasta')
    try:
        with os.fdopen(fd, 'w') as f:
            for sid, seq in sequences.items():
                f.write(f">{sid}\n{seq}\n")
        yield path
    finally:
        os.unlink(path)
```

### GeneAnnotation dataclass

**Goal:** model a genomic interval (gene/feature) as a comparable, sortable object without boilerplate `__init__`/`__eq__`/`__lt__`.

**Approach:** use `@dataclass(order=True)` and mark non-key fields `compare=False` so sorting/equality is based only on genomic position.

```python
from dataclasses import dataclass, field


@dataclass(order=True)
class GeneAnnotation:
    """A genomic feature comparable/sortable by (chromosome, start, end)."""
    chromosome: str
    start: int
    end: int
    name: str = field(compare=False, default="")
    strand: str = field(compare=False, default='+')
    gene_type: str = field(compare=False, default="protein_coding")

    @property
    def length(self) -> int:
        return self.end - self.start

    def overlaps(self, other: "GeneAnnotation") -> bool:
        return (self.chromosome == other.chromosome
                and self.start < other.end
                and other.start < self.end)
```

### SQL — schema, seed data, and bio queries

**Goal:** load gene/variant/expression tables into SQLite and answer real bio questions with joins and aggregation.

**Approach:** build the schema with `executescript`, load rows with `executemany` (never string-format values into SQL), then query with `pd.read_sql_query` for tabular results.

```python
import sqlite3
import pandas as pd


def build_demo_db() -> sqlite3.Connection:
    """Create an in-memory SQLite DB with genes/variants/expression tables."""
    conn = sqlite3.connect(':memory:')
    conn.executescript('''
        CREATE TABLE genes (
            gene_id INTEGER PRIMARY KEY, symbol TEXT, chromosome TEXT,
            start_pos INTEGER, end_pos INTEGER, strand TEXT, biotype TEXT
        );
        CREATE TABLE expression (
            expr_id INTEGER PRIMARY KEY, gene_id INTEGER REFERENCES genes(gene_id),
            sample_id TEXT, tissue TEXT, tpm REAL, condition TEXT
        );
        CREATE TABLE variants (
            variant_id INTEGER PRIMARY KEY, gene_id INTEGER REFERENCES genes(gene_id),
            position INTEGER, ref_allele TEXT, alt_allele TEXT, clinical_significance TEXT
        );
    ''')
    return conn


def genes_with_pathogenic_and_high_tumor_expression(conn: sqlite3.Connection) -> pd.DataFrame:
    """Genes highly expressed in tumor samples (avg TPM > 50) AND carrying a pathogenic variant."""
    return pd.read_sql_query("""
        SELECT symbol FROM genes
        WHERE gene_id IN (
            SELECT gene_id FROM expression WHERE condition = 'tumor'
            GROUP BY gene_id HAVING AVG(tpm) > 50
        ) AND gene_id IN (
            SELECT gene_id FROM variants WHERE clinical_significance = 'pathogenic'
        )
    """, conn)


def variants_for_gene(conn: sqlite3.Connection, symbol: str) -> pd.DataFrame:
    """Look up variants for a gene using a parameterized query (safe against SQL injection)."""
    return pd.read_sql_query(
        """SELECT v.* FROM variants v JOIN genes g ON g.gene_id = v.gene_id
           WHERE g.symbol = ?""",
        conn, params=(symbol,),
    )


# Other common bio queries against the same schema:
# pd.read_sql_query("SELECT symbol, (end_pos-start_pos) AS length FROM genes "
#                    "WHERE (end_pos-start_pos) > 100000 ORDER BY length DESC", conn)
# pd.read_sql_query("SELECT tissue, condition, ROUND(AVG(tpm),2) AS avg_tpm, COUNT(*) AS n "
#                    "FROM expression GROUP BY tissue, condition", conn)
# pd.read_sql_query("SELECT g.symbol, COUNT(v.variant_id) AS n_variants FROM genes g "
#                    "LEFT JOIN variants v ON g.gene_id = v.gene_id "
#                    "GROUP BY g.symbol ORDER BY n_variants DESC", conn)
```

## Pitfalls

- **Missing `@functools.wraps`**: decorated function loses `__name__` and `__doc__`, breaking introspection and logging.
- **Bare `except:`**: catches `SystemExit` and `KeyboardInterrupt`; always catch specific exception types.
- **`__exit__` returning `True`**: suppresses all exceptions silently — only do this intentionally.
- **`@lru_cache` on instance methods**: caches `self`, leaking instances and preventing garbage collection; use on module-level or static functions only.
- **Stacking decorators**: applied bottom-up — `@timer` above `@validate_sequence` means validation runs first, timer measures the whole stack.
- **SQL `HAVING` vs `WHERE`**: `WHERE` filters rows before grouping; `HAVING` filters after aggregation — using `WHERE AVG(tpm) > 50` is a syntax error.
- **`LEFT JOIN` counts**: use `COUNT(v.variant_id)` (a column), not `COUNT(*)`, so genes with zero matches count as 0, not 1.
- **String-formatting values into SQL**: `f"WHERE symbol = '{symbol}'"` is a SQL-injection and quoting-bug risk — always use parameterized queries (`?` placeholders + `params=`).
- **`raise ... from e`**: preserves the original traceback; omitting `from e` inside an `except` block hides the root cause.
- **Properties without a `_`-prefixed backing attribute**: `self.sequence = value` inside a `sequence` setter recurses infinitely; store to `self._sequence`.

## See Also
- `bio-database-access-entrez-fetch` — pull real gene/variant records before loading them into these tables
- `bio-expression-matrix-counts-ingest` — load real RNA-seq count matrices instead of the toy `expression` table
- `bio-variant-calling-vcf-basics` — parse real VCF records into a `variants`-style table
- `polars` — a faster DataFrame alternative to `pandas` for the same SQL-style joins/aggregations