Skip to content

I/O — DataIterator & GFFWriter

DataIterator

Factory function that returns a streaming iterator over GFF3/GTF input (file path, URL, raw string with from_string=True, or an iterable of Feature objects).

gffbase.iterators.DataIterator 

DataIterator(data, checklines: int = 10, transform=None, force_dialect_check: bool = False, from_string: bool = False, **kwargs) -> _DataIterator

Legacy factory. Returns an iterator yielding Feature.

Dispatches on the input, the way the subclasses below have always described and the way gffutils.DataIterator behaves:

  • from_string=True -- data is the GFF text itself.
  • a URL -- fetched to a temporary file first (_UrlIterator).
  • any other path-like -- read from disk, gzipped or not (_FileIterator).
  • an iterable of Feature / ParsedFeature -- yielded straight back (_FeatureIterator), so a generator can be piped into create_db or FeatureDB.update without being written to a file first.

The dispatch was missing: every input went to _DataIterator, which hands whatever it gets to parse_gff(path). So a URL was opened as a filename and an in-memory feature list raised, while the subclasses that exist to handle both sat unreachable and their docstrings described a behaviour the factory did not have.

Source code in python/gffbase/iterators.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def DataIterator(
    data,
    checklines: int = 10,
    transform=None,
    force_dialect_check: bool = False,
    from_string: bool = False,
    **kwargs,
) -> _DataIterator:
    """Legacy factory. Returns an iterator yielding ``Feature``.

    Dispatches on the input, the way the subclasses below have always
    described and the way `gffutils.DataIterator` behaves:

    * ``from_string=True`` -- `data` is the GFF text itself.
    * a URL -- fetched to a temporary file first (`_UrlIterator`).
    * any other path-like -- read from disk, gzipped or not (`_FileIterator`).
    * an iterable of `Feature` / `ParsedFeature` -- yielded straight back
      (`_FeatureIterator`), so a generator can be piped into `create_db` or
      `FeatureDB.update` without being written to a file first.

    The dispatch was missing: every input went to `_DataIterator`, which
    hands whatever it gets to `parse_gff(path)`. So a URL was opened as a
    filename and an in-memory feature list raised, while the subclasses that
    exist to handle both sat unreachable and their docstrings described a
    behaviour the factory did not have.
    """
    if from_string:
        return _DataIterator(
            data,
            checklines=checklines,
            transform=transform,
            force_dialect_check=force_dialect_check,
            from_string=True,
            **kwargs,
        )

    if isinstance(data, (str, os.PathLike)):
        cls = _UrlIterator if is_url(str(data)) else _FileIterator
        return cls(
            os.fspath(data) if isinstance(data, os.PathLike) else data,
            checklines=checklines,
            transform=transform,
            force_dialect_check=force_dialect_check,
            **kwargs,
        )

    if hasattr(data, "__iter__"):
        return _FeatureIterator(data, transform=transform, **kwargs)

    raise TypeError(
        "DataIterator accepts a path, a URL, GFF text with from_string=True, "
        f"or an iterable of features; got {type(data)!r}"
    )

GFFWriter

gffbase.gffwriter.GFFWriter 

GFFWriter(out: str | PathLike | IOBase, with_header: bool = True, in_place: bool = False)

Write Feature records back to a GFF/GTF file.

Source code in python/gffbase/gffwriter.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    out: str | os.PathLike | io.IOBase,
    with_header: bool = True,
    in_place: bool = False,
):
    self.with_header = with_header
    self.in_place = in_place
    self._opened_path: str | None = None
    self._target_path: str | None = None
    self._fh: IO[str]

    if hasattr(out, "write"):
        self._fh = out  # type: ignore[assignment]
    elif in_place:
        # Atomic write via tempfile, swap on close.
        self._target_path = str(out)
        tmp = tempfile.NamedTemporaryFile(
            mode="w",
            delete=False,
            dir=os.path.dirname(self._target_path) or ".",
            suffix=".gffbase.tmp",
            encoding="utf-8",
        )
        self._fh = tmp.file
        self._opened_path = tmp.name
        tmp.close()
        self._fh = open(self._opened_path, "w", encoding="utf-8")
    else:
        self._opened_path = str(out)
        self._fh = open(self._opened_path, "w", encoding="utf-8")

    if self.with_header:
        self._fh.write("##gff-version 3\n")

write_rec 

write_rec(rec: Feature | str) -> None

Write one record, followed by a newline.

Parameters:

  • rec (Feature | str) –

    A Feature, or a pre-formatted GFF line as a string. A trailing newline on a string is not doubled.

Source code in python/gffbase/gffwriter.py 
71
72
73
74
75
76
77
78
79
80
81
82
def write_rec(self, rec: Feature | str) -> None:
    """Write one record, followed by a newline.

    Args:
        rec: A `Feature`, or a pre-formatted GFF line as a string. A
            trailing newline on a string is not doubled.
    """
    if isinstance(rec, str):
        line = rec.rstrip("\n")
    else:
        line = str(rec)
    self._fh.write(line + "\n")

write_recs 

write_recs(recs: Iterable) -> None

Write many records, in the order given.

Parameters:

  • recs (Iterable) –

    An iterable of Feature objects or GFF line strings.

Source code in python/gffbase/gffwriter.py 
84
85
86
87
88
89
90
91
def write_recs(self, recs: Iterable) -> None:
    """Write many records, in the order given.

    Args:
        recs: An iterable of `Feature` objects or GFF line strings.
    """
    for r in recs:
        self.write_rec(r)

write_gene_recs 

write_gene_recs(db: FeatureDB, gene_id: str | Feature) -> None

Write a gene and its ENTIRE subtree, sorted by start.

Parameters:

  • db (FeatureDB) –

    The FeatureDB to read from.

  • gene_id (str | Feature) –

    The gene, as an id or a Feature.

Source code in python/gffbase/gffwriter.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def write_gene_recs(self, db: FeatureDB, gene_id: str | Feature) -> None:
    """Write a gene and its ENTIRE subtree, sorted by start.

    Args:
        db: The `FeatureDB` to read from.
        gene_id: The gene, as an id or a `Feature`.
    """
    gene = db[gene_id] if isinstance(gene_id, str) else gene_id
    self.write_rec(gene)
    for child in db.children(gene, level=None, order_by="start"):
        self.write_rec(child)

write_mRNA_children 

write_mRNA_children(db: FeatureDB, mrna_id: str | Feature) -> None

Write a transcript and its DIRECT children, sorted by start.

Parameters:

  • db (FeatureDB) –

    The FeatureDB to read from.

  • mrna_id (str | Feature) –

    The transcript, as an id or a Feature.

Source code in python/gffbase/gffwriter.py 
105
106
107
108
109
110
111
112
113
114
115
def write_mRNA_children(self, db: FeatureDB, mrna_id: str | Feature) -> None:
    """Write a transcript and its DIRECT children, sorted by start.

    Args:
        db: The `FeatureDB` to read from.
        mrna_id: The transcript, as an id or a `Feature`.
    """
    mrna = db[mrna_id] if isinstance(mrna_id, str) else mrna_id
    self.write_rec(mrna)
    for child in db.children(mrna, level=1, order_by="start"):
        self.write_rec(child)

write_exon_children 

write_exon_children(db: FeatureDB, exon_id: str | Feature) -> None

Write an exon and its direct children, sorted by start.

Parameters:

  • db (FeatureDB) –

    The FeatureDB to read from.

  • exon_id (str | Feature) –

    The exon, as an id or a Feature.

Source code in python/gffbase/gffwriter.py 
117
118
119
120
121
122
123
124
125
126
127
def write_exon_children(self, db: FeatureDB, exon_id: str | Feature) -> None:
    """Write an exon and its direct children, sorted by start.

    Args:
        db: The `FeatureDB` to read from.
        exon_id: The exon, as an id or a `Feature`.
    """
    exon = db[exon_id] if isinstance(exon_id, str) else exon_id
    self.write_rec(exon)
    for child in db.children(exon, level=1, order_by="start"):
        self.write_rec(child)

close 

close() -> None

Flush, and close only a handle this writer opened.

A stream the caller passed in belongs to the caller. Closing it -- as this used to, and as gffutils still does -- means GFFWriter(sys.stdout) shuts stdout down for the whole process, so anything written afterwards raises ValueError: I/O operation on closed file. It is flushed instead, which is the part that actually matters for the output being complete.

Source code in python/gffbase/gffwriter.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def close(self) -> None:
    """Flush, and close only a handle this writer opened.

    A stream the caller passed in belongs to the caller. Closing it -- as
    this used to, and as gffutils still does -- means
    `GFFWriter(sys.stdout)` shuts stdout down for the whole process, so
    anything written afterwards raises `ValueError: I/O operation on closed
    file`. It is flushed instead, which is the part that actually matters
    for the output being complete.
    """
    if self._fh is not None and not self._fh.closed:
        self._fh.flush()
        if self._opened_path is not None:
            self._fh.close()
    if self.in_place and self._opened_path and self._target_path:
        shutil.move(self._opened_path, self._target_path)

export_sqlite

Serialize a GFFBase DuckDB connection back into a legacy gffutils-compatible SQLite database.

gffbase.sqlite_export.export_sqlite 

export_sqlite(con: DuckDBPyConnection, path: str, force: bool = False) -> str

Write a legacy SQLite .db from the given DuckDB connection.

The result is openable by real gffutils. Because gffutils has no way to represent a discontinuous feature, one is flattened back into the N features gffutils itself would have made -- see _legacy_ids. The grouping is not lost: duplicates records (logical_id, legacy_id) for every segment past the first, which is both what that table means and how a re-import can rediscover it.

Returns the absolute path on success.

Source code in python/gffbase/sqlite_export.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def export_sqlite(con: duckdb.DuckDBPyConnection, path: str, force: bool = False) -> str:
    """Write a legacy SQLite ``.db`` from the given DuckDB connection.

    The result is openable by real gffutils. Because gffutils has no way to
    represent a discontinuous feature, one is flattened back into the N
    features gffutils itself would have made -- see `_legacy_ids`. The grouping
    is not lost: `duplicates` records `(logical_id, legacy_id)` for every
    segment past the first, which is both what that table means and how a
    re-import can rediscover it.

    Returns the absolute path on success.
    """
    if os.path.exists(path):
        if not force:
            raise ValueError(f"{path} already exists; pass force=True to overwrite")
        os.unlink(path)

    sqlite_con = sqlite3.connect(path)
    try:
        sqlite_con.executescript(_LEGACY_SCHEMA)
        legacy = _legacy_ids(con)

        # `segments_all`, not `features`: one row per physical input LINE. A
        # segment-0 row must carry its own coordinates here, not the envelope,
        # or the exported file describes spans the source never contained.
        rows = con.execute(
            """
            SELECT feature_id, seg_idx, seqid, source, featuretype, start, "end",
                   score, strand, frame,
                   CAST(attributes_blob AS VARCHAR) AS attributes,
                   CAST(extra_blob      AS VARCHAR) AS extra
            FROM segments_all
            ORDER BY file_order NULLS LAST, feature_id, seg_idx
            """
        ).fetchall()

        # UCSC bin is computed in Python: DuckDB has no equivalent, and
        # `gffutils.FeatureDB.region(completely_within=True)` filters on it, so
        # getting it wrong makes those queries return nothing at all.
        export_rows = []
        for (
            feature_id,
            seg_idx,
            seqid,
            source,
            featuretype,
            start,
            end,
            score,
            strand,
            frame,
            attributes,
            extra,
        ) in rows:
            export_rows.append(
                (
                    legacy.get(f"{feature_id}\x00{seg_idx}", feature_id),
                    seqid,
                    source,
                    featuretype,
                    start,
                    end,
                    score,
                    strand,
                    frame,
                    attributes or "",
                    extra or "",
                    bin_from_coords(start, end),
                )
            )
        sqlite_con.executemany(
            "INSERT INTO features VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
            export_rows,
        )

        # Closure -> relations(parent, child, level=depth), fanned out over
        # both endpoints' legacy ids. A relation naming only the logical id
        # would dangle against a features table that no longer has that row
        # under that name for every segment.
        by_logical: dict[str, list[str]] = {}
        for key, legacy_id in legacy.items():
            by_logical.setdefault(key.split("\x00", 1)[0], []).append(legacy_id)

        rels = []
        for ancestor, descendant, depth in con.execute(
            "SELECT ancestor, descendant, depth FROM closure"
        ).fetchall():
            for parent in [ancestor, *by_logical.get(ancestor, ())]:
                for child in [descendant, *by_logical.get(descendant, ())]:
                    rels.append((parent, child, depth))
        sqlite_con.executemany("INSERT INTO relations VALUES (?,?,?)", rels)

        # `duplicates` is how a re-import rediscovers the grouping, and it is
        # also exactly what the legacy table means: a row that was renamed to
        # avoid a primary-key collision.
        if legacy:
            sqlite_con.executemany(
                "INSERT INTO duplicates VALUES (?, ?)",
                [(key.split("\x00", 1)[0], legacy_id) for key, legacy_id in legacy.items()],
            )

        # Meta — write the dialect (JSON) + version.
        meta = dict(con.execute("SELECT key, value FROM meta").fetchall())
        sqlite_con.execute(
            "INSERT INTO meta VALUES (?, ?)",
            (meta.get("dialect", json.dumps({"fmt": "gff3"})), "gffbase-export"),
        )

        # Directives.
        dirs = con.execute("SELECT directive FROM directives ORDER BY seq").fetchall()
        sqlite_con.executemany("INSERT INTO directives VALUES (?)", dirs)

        # Autoincrements (typically empty).
        try:
            ai = con.execute("SELECT base, n FROM autoincrements").fetchall()
            if ai:
                sqlite_con.executemany("INSERT INTO autoincrements VALUES (?, ?)", ai)
        except duckdb.Error:
            pass

        sqlite_con.commit()
    finally:
        sqlite_con.close()
    return os.path.abspath(path)

Low-level parser

gffbase.parser.parse_gff 

parse_gff(path: str, *, checklines: int = 10, force_dialect_check: bool = False, force_gff: bool = False, strict: bool = True, validation: str = 'ncbi', engine: str | None = 'auto') -> _Iterator

Parse a GFF3/GTF file (plain text or .gz).

Returns an iterator of ParsedFeature plus .dialect(), .directives() and .warnings accessors.

Parameters

validation : {"ncbi", "gffutils"} Which rule set to apply. "ncbi" (default here) is the full GFF3 specification. "gffutils" is the compatibility profile used by create_db: every rule still runs, but a violation annotates the record instead of rejecting it, because real annotation files break the spec routinely and gffutils reads them anyway. strict : bool What a rejection does. True (default) raises GFFFormatError on the first offending line; False skips it and records it in iterator.warnings. Under validation="gffutils" nothing is rejected, so this only affects lines that cannot be parsed at all.

Source code in python/gffbase/parser.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def parse_gff(
    path: str,
    *,
    checklines: int = 10,
    force_dialect_check: bool = False,
    force_gff: bool = False,
    strict: bool = True,
    validation: str = "ncbi",
    engine: str | None = "auto",
) -> _Iterator:
    """Parse a GFF3/GTF file (plain text or ``.gz``).

    Returns an iterator of ``ParsedFeature`` plus ``.dialect()``,
    ``.directives()`` and ``.warnings`` accessors.

    Parameters
    ----------
    validation : {"ncbi", "gffutils"}
        Which rule set to apply. ``"ncbi"`` (default here) is the full GFF3
        specification. ``"gffutils"`` is the compatibility profile used by
        `create_db`: every rule still runs, but a violation annotates the
        record instead of rejecting it, because real annotation files break
        the spec routinely and gffutils reads them anyway.
    strict : bool
        What a *rejection* does. True (default) raises ``GFFFormatError`` on
        the first offending line; False skips it and records it in
        ``iterator.warnings``. Under ``validation="gffutils"`` nothing is
        rejected, so this only affects lines that cannot be parsed at all.
    """
    eng = _resolve_engine(engine)
    if eng == "rust":
        it = _rust.parse_file(
            path,
            checklines=checklines,
            force_dialect_check=force_dialect_check,
            force_gff=force_gff,
            strict=strict,
            validation=validation,
        )
        return _Iterator(it, native=True)
    it = _pyparser.parse_file(
        path,
        checklines=checklines,
        force_dialect_check=force_dialect_check,
        force_gff=force_gff,
        strict=strict,
        validation=validation,
    )
    return _Iterator(it, native=False)

gffbase.parser.parse_bytes 

parse_bytes(data: bytes, *, checklines: int = 10, force_dialect_check: bool = False, force_gff: bool = False, strict: bool = True, validation: str = 'ncbi', engine: str | None = 'auto') -> _Iterator
Source code in python/gffbase/parser.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def parse_bytes(
    data: bytes,
    *,
    checklines: int = 10,
    force_dialect_check: bool = False,
    force_gff: bool = False,
    strict: bool = True,
    validation: str = "ncbi",
    engine: str | None = "auto",
) -> _Iterator:
    eng = _resolve_engine(engine)
    if eng == "rust":
        it = _rust.parse_bytes(
            data,
            checklines=checklines,
            force_dialect_check=force_dialect_check,
            force_gff=force_gff,
            strict=strict,
            validation=validation,
        )
        return _Iterator(it, native=True)
    it = _pyparser.parse_bytes(
        data,
        checklines=checklines,
        force_dialect_check=force_dialect_check,
        force_gff=force_gff,
        strict=strict,
        validation=validation,
    )
    return _Iterator(it, native=False)

gffbase.parser.detect_dialect 

detect_dialect(path: str, *, checklines: int = 10, engine: str | None = 'auto') -> dict
Source code in python/gffbase/parser.py 
206
207
208
209
210
def detect_dialect(path: str, *, checklines: int = 10, engine: str | None = "auto") -> dict:
    eng = _resolve_engine(engine)
    if eng == "rust":
        return _rust.detect_dialect(path, checklines=checklines)
    return _pyparser.detect_dialect(path, checklines=checklines)

gffbase.parser.native_available 

native_available() -> bool

True if the compiled extension is importable.

Source code in python/gffbase/parser.py 
62
63
64
def native_available() -> bool:
    """True if the compiled extension is importable."""
    return _NATIVE