Fyrnheim

0.14.0 verified Sat May 09 auth: no python

Fyrnheim is a typed Python entity framework and dbt alternative built on Pydantic + Ibis. It allows defining typed entities with Pydantic, generating transformations, and running them anywhere. Current version 0.14.0, active development.

pip install fyrnheim

Common errors

error AttributeError: module 'fyrnheim' has no attribute 'derive_entity' ↓

cause API changed; derive functionality moved to Entity.transform()

fix

Use Entity.transform() instead of derive_entity().

error ibis.common.exceptions.IbisError: Expression type 'Table' is not supported ↓

cause Passing a pandas DataFrame directly to Ibis operations without conversion

fix

Use ibis.memtable(df) to convert pandas DataFrames to Ibis tables before passing to Fyrnheim.

Warnings

deprecated The old `Entity.derive()` method has been deprecated in favor of the new transformation API using `Entity.transform()`. ↓

fix Replace `derive()` calls with `transform()`.

gotcha Event source loading now composes Ibis expressions and uses UNION ALL. Previously, it materialized sources locally. If you relied on pandas-only behavior, your pipeline may break. ↓

fix Ensure your data sources are accessible from the Ibis backend (e.g., BigQuery, DuckDB).

breaking The `Ingestor` class signature changed: `target` parameter now expects a connection string instead of a SQLAlchemy engine. ↓

fix Pass a connection string (e.g., 'sqlite:///db.sqlite') instead of an engine object.

Imports

Entity
```
from fyrnheim import Entity
```
Ingestor
```
from fyrnheim import Ingestor
```
EventSource
```
from fyrnheim import EventSource
```
Direct import from fyrnheim, not fyrnheim.events

Quickstart

Defines a User entity and ingests a pandas DataFrame into a database via Ingestor.

import os
import pandas as pd
from fyrnheim import Entity, Ingestor

class User(Entity):
    id: int
    name: str
    email: str

df = pd.DataFrame([
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
])
ingestor = Ingestor(target=os.environ.get('DATABASE_URL', 'sqlite://'))
ingestor.ingest(User, df)
print("Ingestion complete")