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

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.

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.

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

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

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

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.