fastasma

A modular toolkit for FASTA file processing, built around a pipeline of composable operations.

Installation

pip install fastasma

Core concepts

All modules implement the SequenceSource protocol — an iterable that yields Sequence objects (header, sequence, annotation). This allows arbitrary chaining:

output = LastStep(Step2(Step1(ImportFasta("input.fasta"))))

CLI

fastasma -i input.fasta -o output.fasta -p pipeline.yaml

# Multiple pipelines applied in order
fastasma -i input.fasta -o output.tsv -p clean.yaml -p annotate.yaml

# Auto-detects output format by extension (.fasta, .tsv, .db)
fastasma -i input.fasta -o results.tsv -p pipeline.yaml

YAML pipelines

Define reusable, shareable workflows in YAML. Each step is a registered class name with its keyword arguments.

# pipeline.yaml
steps:
  - DropAnnotationKeys:
      keys_to_remove: [length, annotated]

  - Group:
      filter:
        And:
          filters:
            - SequenceLength: {is_: greater, length: 200}
            - ContainsMotif: {motif: ATTG}
      do:
        - DeduplicateHeaders: {separator: _}
        - AddAnnotationToHeader: {annotation_key: organism, separator: _, position: suffix}

  - Sample: {k: 50, seed: 42}

Compound filters in YAML

# Negation
Not:
  filter_:
    ContainsMotif: {motif: TAG}

# Logical AND (all must match)
And:
  filters:
    - ContainsMotif: {motif: ATTG}
    - SequenceLength: {is_: less, length: 300}

# Logical OR (any must match)
Or:
  filters:
    - HasAnnotation: {key: organism, value: Homo sapiens}
    - HeaderMatches: {pattern: "^seq\d+"}

Group without transforms

steps:
  - Group:
      filter:
        Or:
          filters:
            - ContainsMotif: {motif: ATTG}
            - ContainsMotif: {motif: CGGT}

When no do block is given, Group returns only matched sequences (equivalent to .matched).

Nested groups

steps:
  - Group:
      filter:
        SequenceLength: {is_: greater, length: 100}
      do:
        - Group:
            filter:
              ContainsMotif: {motif: ATTG}
            do:
              - DeduplicateHeaders: {}

Python API

Quick start

import fastasma

source = fastasma.ImportFasta("input.fasta")
source = fastasma.DropAnnotationKeys(source, keys_to_remove=["length"])
source = fastasma.AddAnnotationToHeader(source, annotation_key="organism", position="suffix")
source = fastasma.Head(source, n=10)
fastasma.WriteFasta(source, output_path="output.fasta")

Modules

Importers

Annotators

Transform sequence annotations or headers.

Mutators

Filter, sample, or rename sequences in the stream.

Filters

Boolean conditions testable on a single Sequence.

Filter examples

fastasma.ContainsMotif("ATTG")
fastasma.HasAnnotation("organism", "Homo sapiens")
fastasma.HeaderMatches(r"^seq")
fastasma.SequenceLength(is_="greater", length=200)
fastasma.And(fastasma.ContainsMotif("ATTG"), fastasma.SequenceLength(is_="less", length=100))
fastasma.Not(fastasma.ContainsMotif("TAG"))

Groups

Apply operations selectively to matched sequences, preserving original order.

result = (fastasma.Group(fastasma.ImportFasta("input.fasta"), fastasma.ContainsMotif("ATTG"))
          .then(fastasma.AddAnnotationToHeader, annotation_key="organism",
                separator="_", position="suffix")
          .ungroup())
fastasma.WriteFasta(result, output_path="output.fasta")

Registry

Use @register("Name") to make custom classes available in YAML pipelines.

from fastasma.Registry import register
from fastasma.Types import Sequence, SequenceSource

@register("ReverseSequence")
class ReverseSequence:
    def __init__(self, source):
        self._source = source
    def __iter__(self):
        return self.yield_sequence()
    def yield_sequence(self):
        for seq in self._source:
            yield Sequence(seq.header, seq.sequence[::-1], seq.annotation)

steps:
  - ReverseSequence: {}

Writers