API Reference

Assets

class fwirl.Asset(key, dependencies, resources=None, group=None, subgroup=None, allow_retry=True, properties=None)[source]

Bases: object

Abstract base class for a single node in an AssetGraph.

Subclass Asset and implement the timestamp() and build() coroutines to define how fwirl should check whether an asset is up-to-date and how to (re)build it.

Parameters:

  • key – A unique string identifier for this asset.

  • dependencies – A list of Asset objects that must be up-to-date before this asset can be built.

  • resources – Optional list of Resource objects that will be initialized before and cleaned up after every build.

  • group – Optional grouping label used for display purposes.

  • subgroup – Optional sub-grouping label used for display purposes.

  • allow_retry – When True (the default) a failed asset will be retried on the next build pass; when False a failed asset stays in the Failed state until manually unpaused.

  • properties – Optional dictionary of JSON-serializable metadata exposed by the UI.

Example:

import fwirl
import pendulum as plm

class MyAsset(fwirl.Asset):
    async def timestamp(self):
        # Return the time the asset was last produced, or
        # AssetStatus.Unavailable if it does not yet exist.
        ...

    async def build(self):
        # Produce the asset.
        ...
get_key()[source]

Return the string key that uniquely identifies this asset.

Returns:

str – The asset’s key.

abstractmethod async timestamp()[source]

Return the timestamp of the most recently produced asset.

Returns:

A pendulum.DateTime representing when the asset was last built, or AssetStatus.Unavailable if the asset does not yet exist.

abstractmethod async build()[source]

Produce (or re-produce) this asset.

This coroutine is called by fwirl whenever the asset is stale or unavailable. Any return value is ignored; indicate failure by raising an exception.

class fwirl.ExternalAsset(key, dependencies, min_polling_interval, resources=None, group=None, subgroup=None, allow_retry=True)[source]

Bases: Asset

An Asset whose value is managed by an external system.

Use ExternalAsset when the underlying data can change at any time outside of fwirl’s control (for example, a file written by another process, or a row in a database updated by a third-party service). fwirl polls the value at most once every min_polling_interval; if the polled value differs from the cached value the asset is automatically marked stale so that downstream assets are rebuilt.

Subclasses must implement get() (to retrieve the current value) and diff() (to decide whether the new value counts as a change).

Parameters:

  • key – A unique string identifier for this asset.

  • dependencies – A list of Asset objects that must be up-to-date before this asset can be built.

  • min_polling_interval – A pendulum.Duration specifying the minimum time between consecutive polls of the external resource.

  • resources – Optional list of Resource objects.

  • group – Optional grouping label.

  • subgroup – Optional sub-grouping label.

  • allow_retry – See Asset.

async timestamp()[source]

Poll the external resource and return the cached timestamp.

If the minimum polling interval has elapsed, get() is called. When diff() reports a change the asset is marked stale and the new value is staged for the next build() call.

Returns:

A pendulum.DateTime of the last accepted value, or AssetStatus.Unavailable if no value has been accepted yet.

async build()[source]

Commit any pending external value and update the internal timestamp.

abstractmethod async get()[source]

Retrieve the current value of the external resource.

Returns:

The current value in whatever form is appropriate for diff() to compare against.

abstractmethod diff(val)[source]

Return True if val differs from the cached value.

Parameters:

val – The value returned by the most recent get() call.

Returns:

bool – True when the asset should be considered stale.

class fwirl.AssetStatus(value)[source]

Bases: Enum

Enumeration of possible states for an Asset.

Current

The asset is up-to-date with respect to all its dependencies.

Stale

The asset exists but is out-of-date and needs to be rebuilt.

Building

The asset is currently being built.

Paused

The asset has been manually paused and will not be built.

UpstreamStopped

A dependency is paused or failed, blocking this asset.

Unavailable

The asset has not yet been built or does not exist.

Failed

The most recent build attempt for this asset failed.

Current = 0
Stale = 1
Building = 2
Paused = 3
UpstreamStopped = 4
Unavailable = 5
Failed = 6

Asset Graph

class fwirl.AssetGraph(key, notifiers=None)[source]

Bases: object

A directed acyclic graph of Asset objects with a built-in scheduler and messaging interface.

AssetGraph is the central object in fwirl. It holds a DAG of assets, tracks their statuses, and orchestrates builds and refreshes both on demand and on a cron-based schedule. It can also receive commands at runtime through RabbitMQ by calling run().

Parameters:

  • key – A unique string identifier for this graph. The same key is used as the RabbitMQ routing key when running in server mode.

  • notifiers –

    Optional mapping of notifiers service configurations used to send build notifications. Each entry has the form:

    {
    "<service_name>": {
        "params": {...},   # notifiers service params
        "level": "ERROR",  # loguru level string
    }
}

Example:

import fwirl

g = fwirl.AssetGraph("my_graph")
g.add_assets([asset_a, asset_b])
g.schedule("hourly_build", "build", "0 * * * *")
g.run()  # blocks; listens for RabbitMQ commands
add_assets(assets)[source]

Add a list of assets (and all their transitive dependencies) to the graph.

The method walks the dependency tree rooted at each asset in assets and inserts every discovered node and edge into the internal DiGraph. Duplicate assets (same key) are silently skipped.

Parameters:

assets – An iterable of Asset objects to add.

Raises:

ValueError – If a self-dependency loop is detected, or if any asset’s build() or timestamp() method is not a coroutine.

remove_assets(assets)[source]

Remove a list of assets from the graph.

Downstream assets that depended on any removed node are also removed automatically by NetworkX.

Parameters:

assets – An iterable of Asset objects to remove.

list_assets(display=True)[source]

Return (and optionally log) a formatted list of all assets.

Parameters:

display – When True (the default) the string is also emitted via the loguru logger.

Returns:

str – A multi-line string listing each asset key and its current AssetStatus.

list_schedules(display=True)[source]

Return (and optionally log) a formatted list of all schedules.

Parameters:

display – When True (the default) the string is also emitted via the loguru logger.

Returns:

str – A multi-line string listing each schedule name and its Schedule representation.

list_jobs(display=True)[source]

Return (and optionally log) the current job queue.

Parameters:

display – When True (the default) the string is also emitted via the loguru logger.

Returns:

str – A multi-line string describing the currently running job (if any) and the pending job queue.

update_asset_properties(asset_key, properties)[source]
schedule(schedule_key, action, cron_string='', asset=None, immediate_once=False)[source]

Add a recurring (or one-shot) schedule to the graph.

Parameters:

  • schedule_key – A unique string identifier for the schedule.

  • action – "refresh" to trigger a status refresh, or "build" to trigger a full build.

  • cron_string – A standard cron expression (e.g. "0 6 * * *" for 06:00 every day). Ignored when immediate_once is True.

  • asset – Optional Asset targeted by the action. When None the action applies to all assets.

  • immediate_once – When True the schedule fires once as soon as possible and is then automatically removed.

unschedule(schedule_key)[source]

Remove a schedule from the graph.

Parameters:

schedule_key – The string key of the schedule to remove.

pause_schedule(schedule_key)[source]

Pause a schedule so that it no longer fires.

Parameters:

schedule_key – The string key of the schedule to pause.

pause_asset(asset)[source]

Set an asset’s status to Paused.

A paused asset is skipped during builds and its downstream assets are marked UpstreamStopped.

Parameters:

asset – The Asset to pause.

set_allow_retry(asset, allow_retry)[source]

Set whether a failed asset is retried on the next build pass.

See the allow_retry argument of Asset.

Parameters:

  • asset – The Asset to update.

  • allow_retry – True to retry failures automatically, False to keep the asset Failed until it is manually unpaused.

unpause_schedule(schedule_key)[source]

Resume a previously paused schedule.

Parameters:

schedule_key – The string key of the schedule to resume.

unpause_asset(asset)[source]

Mark a paused asset as Stale so it will be rebuilt on the next build pass.

Parameters:

asset – The Asset to unpause.

run()[source]

Start the graph server and block until a shutdown signal is received.

Launches a background thread that listens for RabbitMQ messages addressed to this graph’s key, and runs the async job scheduler in the main thread. The method returns when:

  • a shutdown message is received via RabbitMQ, or

  • the user presses Ctrl-C.

Typical usage:

g = fwirl.AssetGraph("my_graph")
g.add_assets([...])
g.schedule("hourly", "build", "0 * * * *")
g.run()
refresh(assets=None)[source]

Synchronously refresh the status of all (or specified) assets.

Checks each asset’s timestamp() against its dependencies and updates status accordingly. Does not call build().

Parameters:

assets – A single Asset or list of assets to refresh. When None (the default) every asset in the graph is refreshed.

build(assets=None)[source]

Synchronously build all (or specified) stale and unavailable assets.

Refreshes statuses first, then calls build() on every asset that is not current.

Parameters:

assets – A single Asset or list of assets to build (fwirl will also build all upstream dependencies). When None (the default) all assets are considered.

summarize(display=True)[source]

Return (and optionally log) a human-readable status summary.

The summary includes total asset and edge counts broken down by AssetStatus, grouped by the group/subgroup labels assigned to each asset.

Parameters:

display – When True (the default) the summary is also emitted via the loguru logger.

Returns:

str – The formatted summary string.

Resources

class fwirl.resource.Resource(key)[source]

Bases: object

Abstract base class for a managed build resource.

Resources represent shared, stateful objects (database connections, file handles, hardware locks, etc.) that must be acquired before a build begins and released when it finishes. Attach one or more Resource instances to an Asset via its resources constructor argument; fwirl will call init() before the build pass and close() after it completes (or fails).

Parameters:

key – A unique string identifier for this resource.

Example:

import fwirl

class DBConnection(fwirl.Resource):
    def init(self):
        self.conn = connect_to_db()

    def close(self):
        self.conn.close()
abstractmethod init()[source]

Acquire / open the resource.

Called once before the build pass that needs this resource begins. Raise an exception to abort the build.

abstractmethod close()[source]

Release / close the resource.

Called after every build pass, even when init() raised an exception for a different resource in the same build.

Graph Workers

class fwirl.worker.GraphWorker(watched_assets)[source]

Bases: object

Abstract base class for dynamic graph restructuring callbacks.

Attach a GraphWorker to an AssetGraph to receive a callback after any of its watched_assets are successfully rebuilt. Implement restructure() to inspect the graph and return new assets that should be added before fwirl triggers another build pass.

This is useful when the set of assets that must be maintained is not known up front but is determined by the content of upstream assets (for example, discovering a list of files to process from a manifest asset).

Parameters:

watched_assets – An iterable of Asset objects whose rebuild should trigger restructure().

abstractmethod async restructure(graph)[source]

Inspect the graph and optionally add new assets.

This coroutine is called after any of the watched assets finish rebuilding. Use the graph parameter to add new assets and return them so that fwirl can immediately schedule a build for the new nodes.

Parameters:

graph – The live AssetGraph instance.

Returns:

A list of newly added Asset objects, or an empty list if no new assets were added.

Schedules

class fwirl.schedule.Schedule(name, func, kwargs, cron_string='* * * * *', immediate_once=False)[source]

Bases: object

A named, optionally-recurring job attached to an AssetGraph.

Schedule objects are created internally by AssetGraph.schedule and are not normally constructed directly by user code.

Parameters:

  • name – A unique string identifier for this schedule.

  • func – The async callable to invoke when the schedule fires.

  • kwargs – Keyword arguments passed to func on each invocation.

  • cron_string – A standard cron expression ("* * * * *" by default) that controls when the schedule fires. Ignored when immediate_once is True.

  • immediate_once – When True the schedule fires once as soon as possible and is then removed from the graph.

IMMEDIATE = DateTime(1970, 1, 1, 0, 0, 0, tzinfo=Timezone('UTC'))
next(dt=None)[source]

Return seconds until the next scheduled firing, or IMMEDIATE.

Parameters:

dt – Reference pendulum.DateTime; defaults to now.

Returns:

Number of seconds (float) until the next occurrence, or IMMEDIATE for one-shot schedules.

pause()[source]

Prevent this schedule from firing until unpause() is called.

unpause()[source]

Allow this schedule to fire again after a pause().

is_paused()[source]

Return True if the schedule is currently paused.

Returns:

bool – Paused state.

generate_coroutine()[source]

Create and return the coroutine for the next scheduled execution.

Returns:

A coroutine object ready to be awaited.

Remote API

fwirl.api.list_graphs discovers running graphs on the local machine. The remaining functions communicate with a running AssetGraph server over RabbitMQ. They are also exposed as CLI commands.

fwirl.api.summarize(graph_key, rabbit_url='amqp://guest:guest@localhost//')[source]

Print a summary of the running asset graph to stdout.

Sends a summarize message to the graph server identified by graph_key and waits for its response.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • rabbit_url – RabbitMQ connection URL. Defaults to ******localhost//.

fwirl.api.list_graphs()[source]

List the graph keys of currently running asset graph servers.

Returns:

list[str] – Sorted graph keys discovered on the local machine.

fwirl.api.ls(graph_key, assets=False, schedules=False, jobs=False, rabbit_url='amqp://guest:guest@localhost//')[source]

List the contents of a running asset graph.

Sends an ls message to the graph server and prints the response. Pass one or more flags to control which information is included.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • assets – When True, include the list of assets and their statuses.

  • schedules – When True, include the list of schedules.

  • jobs – When True, include the job queue.

  • rabbit_url – RabbitMQ connection URL.

fwirl.api.build(graph_key, asset_key=None, rabbit_url='amqp://guest:guest@localhost//')[source]

Trigger a build on a running asset graph.

Rebuilds all stale or unavailable assets (or the specified asset and its upstream dependencies).

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • asset_key – Optional string key of a specific asset to build. When None (the default) all assets are built.

  • rabbit_url – RabbitMQ connection URL.

fwirl.api.refresh(graph_key, asset_key=None, rabbit_url='amqp://guest:guest@localhost//')[source]

Trigger a status refresh on a running asset graph.

A refresh checks the timestamps of every asset (or just the specified asset and its downstream dependents) without rebuilding any of them.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • asset_key – Optional string key of a specific asset to refresh. When None (the default) all assets are refreshed.

  • rabbit_url – RabbitMQ connection URL.

fwirl.api.pause(graph_key, key=None, rabbit_url='amqp://guest:guest@localhost//')[source]

Pause an asset or schedule in a running asset graph.

A paused asset will not be rebuilt until it is unpaused. A paused schedule will not fire until it is unpaused.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • key – String key of the asset or schedule to pause.

  • rabbit_url – RabbitMQ connection URL.

fwirl.api.unpause(graph_key, key=None, rabbit_url='amqp://guest:guest@localhost//')[source]

Resume a previously paused asset or schedule.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • key – String key of the asset or schedule to unpause.

  • rabbit_url – RabbitMQ connection URL.

fwirl.api.schedule(graph_key, schedule_key, action, cron_string, asset_key=None, rabbit_url='amqp://guest:guest@localhost//')[source]

Add a new recurring schedule to a running asset graph.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • schedule_key – A unique string identifier for the new schedule.

  • action – The action to perform; either "refresh" or "build".

  • cron_string – A standard cron expression controlling when the schedule fires (e.g. "0 * * * *" for every hour).

  • asset_key – Optional string key of the asset targeted by the action. When None the action applies to all assets.

  • rabbit_url – RabbitMQ connection URL.

fwirl.api.unschedule(graph_key, schedule_key, rabbit_url='amqp://guest:guest@localhost//')[source]

Remove an existing schedule from a running asset graph.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • schedule_key – The string key of the schedule to remove.

  • rabbit_url – RabbitMQ connection URL.

fwirl.api.shutdown(graph_key, rabbit_url='amqp://guest:guest@localhost//')[source]

Send a shutdown signal to a running asset graph server.

Parameters:

  • graph_key – The string key of the running AssetGraph.

  • rabbit_url – RabbitMQ connection URL.