# Python Bio Context Managers > Build Python context managers (__enter__/__exit__, @contextmanager, sqlite3) for safe FASTA I/O, temp cleanup, DB transactions. Use for leaked file handles, temp files surviving crashes, or with-compatible readers/writers. - Skill: `pavel-kravchenko/python-bio-context-managers` (Agent Skill) - Install (CLI): `npx skillmds@latest add pavel-kravchenko/python-bio-context-managers` - Raw SKILL.md: https://api.skillmd.com/api/skills/pavel-kravchenko/python-bio-context-managers/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - Author: pavel-kravchenko (https://skillmd.com/u/pavel-kravchenko) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/pavel-kravchenko/python-bio-context-managers --- # Context Managers for Bioinformatics ## When to Use - A pipeline crashes mid-run and leaves open file handles, locked SQLite databases, or orphaned temp files. - You need a `with`-compatible FASTA/VCF/BAM writer or reader that always closes on exit, even on exception. - You want auto-cleanup of a temporary file (e.g., a scratch FASTA for a subprocess call) regardless of success/failure. - You're wrapping a multi-step database insert (variant calls, sample metadata) that must commit atomically or roll back. - You want to time or instrument a block of pipeline code without cluttering it with try/finally boilerplate. ## Version Compatibility Pure standard library — `contextlib` and `sqlite3` are part of CPython. Applies to Python ≥3.8 (3.10+ recommended for `dict[str, str]`-style built-in generics used below). ## Prerequisites - No third-party packages required (stdlib only: `contextlib`, `tempfile`, `sqlite3`, `os`, `time`). - Familiarity with Python classes and generators helps (`__enter__`/`__exit__`, `yield`). - Related: `bio-sequence-io-write-sequences`, `bio-sequence-io-read-sequences` for FASTA I/O without a custom context manager. ## Lifecycle ```text with ctx as resource: __enter__() → acquire (body) __exit__() → release (always runs, even on exception) ``` `__exit__(exc_type, exc_val, tb)` — return `True` to suppress the exception, `False`/`None` to re-raise. ## Class-Based Context Manager **Goal:** build a `with`-compatible FASTA writer that always closes its file handle, even if the caller's code raises mid-write. **Approach:** implement `__enter__` (acquire) and `__exit__` (release, never suppress), track state as instance attributes. ```python class FastaWriter: """Context manager for writing line-wrapped FASTA records.""" def __init__(self, filename: str, line_width: int = 80): self.filename = filename self.line_width = line_width self.file = None self.record_count = 0 def __enter__(self): self.file = open(self.filename, 'w') return self def __exit__(self, exc_type, exc_val, tb): if self.file: self.file.close() if exc_type is not None: print(f"Error occurred while writing: {exc_val}") print(f"Wrote {self.record_count} records to {self.filename}") return False # never suppress exceptions def write_record(self, seq_id: str, sequence: str, description: str = "") -> None: """Write one FASTA record, wrapping sequence lines to self.line_width.""" header = f">{seq_id}" + (f" {description}" if description else "") self.file.write(header + "\n") for i in range(0, len(sequence), self.line_width): self.file.write(sequence[i:i + self.line_width] + "\n") self.record_count += 1 with FastaWriter("output.fasta") as writer: writer.write_record("BRCA1", "ATGGATTTCGATCG" * 10, "breast cancer gene") ``` ## `@contextmanager` (generator style) **Goal:** auto-delete a scratch FASTA file after use (e.g., input to a subprocess aligner call), regardless of success or failure. **Approach:** code before `yield` is setup, code after is teardown — always wrap teardown in `try/finally` so cleanup runs even if the body raises. ```python from contextlib import contextmanager import tempfile import os @contextmanager def temp_fasta(sequences: dict[str, str]): """Write sequences to a temp FASTA file, yield its path, delete on exit.""" fd, path = tempfile.mkstemp(suffix=".fasta") try: with os.fdopen(fd, 'w') as f: for name, seq in sequences.items(): f.write(f">{name}\n{seq}\n") yield path finally: os.unlink(path) with temp_fasta({"seq1": "ATGC", "seq2": "TTAA"}) as path: # path is valid here; deleted after the block, even on exception with open(path) as f: print(f.read()) ``` ## Timed Section + Stats-Tracking Processor **Goal:** time an expensive block, and separately track running stats (records parsed, bases seen) across a FASTA parse without polluting the parsing loop with timing code. **Approach:** a `@contextmanager` for ad-hoc timing; a class-based manager when you need both iteration (a generator method) and end-of-run summary stats in `__exit__`. ```python import time from contextlib import contextmanager from dataclasses import dataclass @contextmanager def timed_section(name: str): """Print elapsed wall-clock time for the wrapped block.""" start = time.perf_counter() try: yield finally: elapsed = time.perf_counter() - start print(f"[{name}] {elapsed:.4f}s") @dataclass class FastaRecord: id: str description: str sequence: str @property def gc_content(self) -> float: seq = self.sequence.upper() return (seq.count('G') + seq.count('C')) / len(seq) * 100 class FastaProcessor: """Context manager that parses FASTA and reports summary stats on exit.""" def __init__(self, filename: str): self.filename = filename self.file = None self.records_processed = 0 self.total_bases = 0 def __enter__(self): self.file = open(self.filename, 'r') return self def __exit__(self, exc_type, exc_val, tb): if self.file: self.file.close() print(f"Records: {self.records_processed}, total bases: {self.total_bases:,}") return False def records(self): """Yield FastaRecord objects one at a time (streaming, low memory).""" current_id, current_desc, parts = None, "", [] for line in self.file: line = line.strip() if line.startswith('>'): if current_id: seq = ''.join(parts) self.records_processed += 1 self.total_bases += len(seq) yield FastaRecord(current_id, current_desc, seq) header_parts = line[1:].split(None, 1) current_id = header_parts[0] current_desc = header_parts[1] if len(header_parts) > 1 else "" parts = [] elif current_id: parts.append(line) if current_id: seq = ''.join(parts) self.records_processed += 1 self.total_bases += len(seq) yield FastaRecord(current_id, current_desc, seq) with timed_section("parse FASTA"): with FastaProcessor("output.fasta") as processor: for record in processor.records(): print(f"{record.id}: {len(record.sequence)} bp, GC={record.gc_content:.1f}%") ``` ## SQLite Transactions ```python import sqlite3 with sqlite3.connect("variants.db") as conn: # auto-commits on success, rolls back on exception conn.execute("INSERT INTO variants VALUES (?, ?, ?)", ("chr1", 100, "A")) ``` ## Multi-resource `with` ```python with open("genome.fasta") as fasta, open("variants.vcf") as vcf: # both files open; both closed on exit, inner (vcf) closes first ... ``` ## Pitfalls - **`yield` position matters**: code after `yield` is teardown — an exception in the body skips cleanup unless wrapped in `try/finally`. - **`__exit__` receives the exception, not `__enter__`**: if the `with` block raises, Python calls `__exit__` with exception info; forgetting to return `False` can accidentally suppress exceptions. - **`with conn:` on SQLite commits, not closes**: `conn.close()` is separate; use `with contextlib.closing(conn):` if you also want auto-close. - **Nesting vs stacking**: `with A() as a, B() as b:` is equivalent to two nested `with` statements; `B.__exit__` runs before `A.__exit__`. - **Reusable vs reentrant**: a class-based context manager instance is reusable across separate `with` blocks by default, but is *not* reentrant (don't nest the same instance inside itself) unless you explicitly design for it. ## See Also - `bio-sequence-io-write-sequences` — writing FASTA/FASTQ without a custom context manager (Biopython `SeqIO.write`). - `bio-sequence-io-read-sequences` — streaming FASTA/FASTQ parsing. - `bio-database-access-batch-downloads` — retry/backoff patterns for unreliable network fetches, complementary to resource cleanup here.