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

SQL Clauses

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).

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.

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.

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.

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