Open Data Framework

Lightweight dependency-injection framework for data applications.

This package provides structural abstractions — Entity, Repository, Component, Service, Task, Pipeline, Agent, Function, Layer, Namespace, Context, Project, Config, Logger, Events, Notifications, View — with zero third-party dependencies. No UI, no CLI, no MCP server.

Need the CLI, UI, or MCP server? That's the odf package, which depends on this one.

Full documentation: opendataframework.github.io/opendataframework

---

Why

Most data projects solve the same structural problems repeatedly: how to wire dependencies, manage component lifecycles, and keep configuration separate from code. This framework provides that plumbing as a small set of reusable abstractions — Entity, Repository, Component, Service, Task, Pipeline — so it doesn't need to be reinvented by hand in every project.

Want a ready-made project layout instead? Scaffolding an actual project from a template is the odf package's job — this is what that scaffold is built on.

---

Getting Started

Install:

pip install opendataframework

A minimal project needs three things: an entity and repository, a config file, and a main file that starts the Project.

Application code (app.py):

import sqlite3
from dataclasses import dataclass

from opendataframework import Component, Config, Entity, Repository


@Entity
@dataclass
class User:
    id: int
    name: str


@Component
class SQLite:
    def __init__(self, config: Config):
        self.conn = sqlite3.connect(config.component.sqlite.path)
        self.conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")


@Repository(User)
class Users:
    def __init__(self, sqlite: SQLite):
        self.db = sqlite.conn

    def all(self) -> list[User]:
        rows = self.db.execute("SELECT id, name FROM users").fetchall()
        return [User(id=row[0], name=row[1]) for row in rows]

    def save(self, user: User) -> None:
        self.db.execute("INSERT OR REPLACE INTO users VALUES (?, ?)", (user.id, user.name))
        self.db.commit()

Config file (config.toml):

[component.sqlite]
path = "app.db"

Main file (main.py):

from opendataframework import Project

from app import User, Users

project = Project.from_config("config.toml")
project.start()

users = project.context.get(Users)
users.save(User(id=1, name="Ada"))

print(users.all())

project.start() blocks until every component has completed its initialisation stage — there is no hidden latency on first access. Adding a new component means declaring its dependencies in the constructor, not writing wiring code — a config section is only needed if the component reads from Config itself.

---

Core Concepts

Entity

A structured unit of data the system works with.

@Entity
@dataclass
class User:
    id: int
    name: str

Repository

Manages Entities; abstracts persistence away from business logic.

@Repository(User)
class Users:
    def get(self, user_id: int) -> User: ...
    def save(self, user: User) -> None: ...

View

An optional, single declaration a Repository attaches to itself: which representation (table, map, image, video, ...) and field(s) best fit its data. Metadata only — this package renders nothing itself. A repository whose view points at content (an image, a clip, a document) may also implement content_metadata() to declare each record's file name and MIME type. See View.

@Repository(Store)
class Stores:
    def all(self) -> list[Store]: ...

    def data_view(self) -> LocationView:
        return LocationView(fields=("lat", "lon"))

Component

A Context-managed object with no required execution contract — wired in and given its dependencies through the constructor.

@Component
class Classifier:
    def fit(self, data): ...
    def predict(self, data): ...

Service

A long-running executable, wired through DI the same way as a Component but with a mandatory setup → run → stop lifecycle — an API server, worker process, or scheduler. run() blocks by nature; the framework backgrounds it automatically.

@Service
class Postgres:
    def setup(self): ...
    def run(self): ...
    def stop(self): ...

Task

A finite executable, wired through DI the same way as a Component — performs one bounded unit of work, once.

@Task
class MetricsFetcher:
    def execute(self): ...

Pipeline

Coordinates multiple Tasks (and optionally other Pipelines) into an ordered workflow.

@Pipeline
class DailyAnalytics:
    def execute(self): ...

Agent

A finite executable whose steps are decided at run time, typically by a language model — handed its goal as arguments to invoke(). What it may act on is declared through its constructor. The framework ships no model client; invoke() is where a library of your choice goes.

@Agent
class Librarian:
    def __init__(self, books: Books, model: Model): ...

    def invoke(self, question: str) -> str: ...


project.context.call("agent.librarian", question="Anything by Herbert?")

Function

A plain function made framework-aware — for adopting existing code without a class refactor first. Returned unchanged; called through the Context, its managed dependencies are injected and the rest are arguments.

@Function
def load_data(path: str, db: Postgres) -> int: ...


project.context.call("function.load-data", path="data.csv")  # db injected

A class asks for a function through its constructor, and gets it with its dependencies already filled in:

@Task
class Refresh:
    def __init__(self, load: Annotated[Callable[..., int], load_data]):
        self.load = load

    def execute(self):
        return self.load("data.csv")

Layer

An organisational grouping — which subsystem a component belongs to (Api, Storage, Analytics, Messaging, Monitoring, Security, or custom). Independent of execution type.

@Storage
@Service
class Postgres: ...

Namespace

The base class underlying every framework decorator — Entity, Component, Service, Task, Pipeline, Agent, Function, Repository, Layer, Subscriber, and Publisher are all Namespace subclasses, each with its own independent name → class mapping. A framework extension point, not something end users subclass directly. See Namespace.

Config

Typed, dot-accessible view over a configuration dict — snake_case and kebab-case are interchangeable, and nested sections chain naturally. Resolved automatically into the container by Project.from_config.

@Storage
@Service
class Postgres:
    def __init__(self, config: Config):
        self.host = config.postgres.host

Logger

Per-component logging handle, injectable via constructor DI like any other dependency — writes land in that component's own log file with no name argument needed. Logs are structured: one JSON record per line by default, keyword arguments become fields, and every protocol call the framework makes is logged with its call and parent ids, so a run can be traced across files. Files can rotate by size (max-bytes and backups under [project.logging]); see Logger.

@Task
class ExportUsers:
    def __init__(self, users: Users, logger: Logger):
        self.users = users
        self.logger = logger

    def execute(self) -> None:
        self.logger.info("export starting", users=len(self.users.all()))

Events

Reports every protocol call the framework makes — a Service's setup(), a Task's execute(), a @Function call — as started, then completed or failed, with timing and the parent call it ran inside.

@Component
class FailureAlerts:
    def __init__(self, events: Events):
        self._unsubscribe = events.subscribe(self.on_event)

    def on_event(self, event: Event) -> None:
        if event.state == "failed":
            ...

Notifications

Lets a component announce that something happened — a file arrived, an order came in — so others can react without the publisher knowing who they are. Never emitted by the framework itself. A publisher declares the topics it publishes, and a subscriber the topics it reacts to; each reaction is its own traced call.

@Publisher("orders.received")
@Task
class TakeOrders:
    def __init__(self, notifications: Notifications):
        self.notifications = notifications

    def execute(self) -> None:
        self.notifications.publish("orders.received", {"id": 7}, summary="Le Petit: 2 x baguette")


@Subscriber("orders.*")  # deliver="background" for a slow reaction
@Component
class Kitchen:
    def on_notification(self, notification: Notification) -> None:
        print("cooking order", notification.payload["id"])

Context

Registers, resolves, and lifecycles every component in dependency order. No circular dependencies — restructure with a shared third collaborator instead.

component = project.context.get(UsersApi)  # typed, by class

Project

The composition root — owns Context and configuration, and is the single entry point for starting and stopping the application.

project = Project.from_config("config.toml")
project.start()

Multiple independent Project instances are supported in the same process; there is no global mutable state.

---

Design Guidelines

• Prefer composition — decorators (@Api @Service), not inheritance from framework base classes.
• Keep responsibilities focused — one primary reason to change per class.
• Avoid global mutable state — framework state belongs to objects.
• Avoid hidden magic — dependencies must be visible in constructor signatures.

---

Mental Model

Data is represented by Entities.
Repositories manage data.

Components provide capabilities.
Services run for extended periods.
Tasks execute bounded work.
Pipelines coordinate tasks (and other pipelines).
Agents pursue a goal handed to them.

Layers organize components.
Context manages components and repositories.
Project owns everything.

---

Examples

See examples/ — small, focused projects each isolating one core abstraction.

For the CLI, UI, MCP server, and chat surface built on top of this package, see odf.

---

Development

Requires Python >=3.14, managed with Poetry.

poetry install       # install dependencies, including dev
poetry run pytest    # run the test suite

See CONTRIBUTING.md for how to propose changes.

---

License

MIT