Skip to content

Python API Reference

This reference is generated from the installed package. Run python scripts/gen_api_md.py to regenerate.

HTTP client for the VoLCA REST API.

Usage::

c = Client(db="agribalyse-3.2", password="1234")
plants = c.search_activities(name="at plant")
chain = c.get_supply_chain(plants[0].process_id, name="at farm")

Substitutions can be passed to get_supply_chain, get_inventory, and get_impacts to compute results with a different upstream supplier — fast::

subs = [{"from": old_pid, "to": new_pid, "consumer": consumer_pid}]
result = c.get_impacts(pid, method_id=mid, substitutions=subs)

Constructor: Client(base_url: str = 'http://localhost:8080', db: str = '', password: str = '')

Methods:

  • aggregate(process_id: str, scope: str, *, is_input: bool | None = None, max_depth: int | None = None, filter_name: str | None = None, filter_name_not: list[str] | str | None = None, filter_unit: str | None = None, preset: str | None = None, filter_classification: list[ClassificationFilter] | None = None, filter_target_name: str | None = None, filter_is_reference: bool | None = None, group_by: str | None = None, aggregate: str | None = None) -> AggregateResult — SQL-group-by aggregation over direct exchanges, supply chain, or biosphere flows.
  • call(operation_id: str, **kwargs) -> Any — Escape hatch: call any OpenAPI operation by operationId.
  • get_activity(process_id: str) -> ActivityDetail — Fetch an activity’s full detail.
  • get_characterization(method_id: str, *, flow: str | None = None, limit: int | None = None) -> dict — Look up characterization factors for a method matched to database flows.
  • get_consumers(process_id: str, *, name: str | None = None, location: str | None = None, product: str | None = None, preset: str | None = None, classification_filters: list[ClassificationFilter] | None = None, limit: int | None = None, max_depth: int | None = None, include_edges: bool = False) -> ConsumersResponse — Find all activities that transitively consume this supplier.
  • get_contributing_activities(process_id: str, method_id: str, *, collection: str = 'methods', limit: int | None = None) -> dict — Which upstream activities drive a given impact category.
  • get_contributing_flows(process_id: str, method_id: str, *, collection: str = 'methods', limit: int | None = None) -> dict — Which elementary flows drive a given impact category.
  • get_flow_mapping(method_id: str) -> dict — Get the characterization-factor-to-database-flow mapping coverage.
  • get_impacts(process_id: str, method_id: str, *, collection: str = 'methods', top_flows: int | None = None, substitutions: list[dict] | None = None) -> LCIAResult — Compute the LCIA score for a single impact category on an activity.
  • get_impacts_batch(process_id: str, *, collection: str = 'methods', substitutions: list[dict] | None = None) -> LCIABatchResult — Compute LCIA for every impact category in a collection, in one call.
  • get_inputs(process_id: str) -> list[Exchange] — Return the input exchanges of an activity (richer metadata than get_activity).
  • get_inventory(process_id: str, *, flow: str | None = None, limit: int | None = None, substitutions: list[dict] | None = None) -> dict — Compute the life-cycle inventory (cumulative biosphere flows) for an activity.
  • get_outputs(process_id: str) -> list[Exchange] — Return the output exchanges of an activity. See :meth:get_inputs for notes.
  • get_path_to(process_id: str, target: str) -> PathResult — Find the shortest upstream path from process to first activity whose name matches target.
  • get_supply_chain(process_id: str, *, name: str | None = None, location: str | None = None, limit: int | None = None, min_quantity: float | None = None, max_depth: int | None = None, preset: str | None = None, classification_filters: list[ClassificationFilter] | None = None, substitutions: list[dict] | None = None, include_edges: bool | None = None) -> SupplyChain — Get the flat supply chain of an activity.
  • get_tree(process_id: str) -> dict — Fetch the recursive activity tree used by the analysis SPA.
  • get_version() — Return server version info (version, gitHash, gitTag, buildTarget).
  • list_classifications() — List classification systems and their values for the current database.
  • list_databases() — List every database declared in the engine config.
  • list_methods() — List every LCIA method available in the engine.
  • list_presets() — List classification presets configured in this instance.
  • load_database(db_name: str) -> dict — Load a database into memory so it answers queries.
  • refresh_stubs() — Fetch the OpenAPI spec from the server and refresh the dispatch table.
  • search_activities(name: str | None = None, *, geo: str | None = None, product: str | None = None, preset: str | None = None, classification: str | None = None, classification_value: str | None = None, limit: int | None = None, offset: int = 0, exact: bool = False) -> list[Activity] — Search activities in the current database.
  • search_flows(query: str | None = None, *, limit: int | None = None) -> list[dict] — Search flows (technosphere products and biosphere flows) in the current database.
  • unload_database(db_name: str) -> dict — Unload a database from memory to free RAM. The disk copy is kept.
  • use(db_name: str) -> 'Client' — Return a new client targeting a different database (shares session).

Manages the VoLCA server process.

Usage::

with Server(config="volca.toml") as srv:
client = Client(base_url=srv.base_url, db="agribalyse-3.2", password=srv.password)
activities = client.search_activities(name="at plant")

Constructor: Server(config: str = 'volca.toml', port: int = 0, binary: str = 'volca')

Properties:

  • base_url — (no docstring)

Methods:

  • is_alive() — Health check — GET /api/v1/db, return True if 200.
  • start(idle_timeout: int = 300, wait_timeout: int = 120) -> None — Spawn the engine process if it is not already serving, and wait until ready.
  • stop() — Stop the server via shutdown endpoint, then terminate process.

Raised when the download or verification fails.

Error from the VoLCA API.

Constructor: VoLCAError(message: str, status_code: int | None = None, body: str = '')

FieldTypeDefault
process_idstr
namestr
locationstr
productstr
product_amountfloat
product_unitstr

Typed wrapper around the JSON returned by GET /activity/{pid}.

Use the .inputs / .outputs / .technosphere_inputs convenience properties instead of walking the raw exchanges list.

FieldTypeDefault
process_idstr
namestr
locationstr
unitstr
descriptionlist[str]
classificationsdict[str, str]
reference_productstr | None
reference_product_amountfloat | None
reference_product_unitstr | None
all_productslist[Activity]
exchangeslist[Union[TechnosphereExchange, BiosphereExchange]]

Result of compare_activities.

FieldTypeDefault
scopestr
group_bystr
matchedlist[ActivityDiffRow]list()
left_onlylist[ActivityDiffRow]list()
right_onlylist[ActivityDiffRow]list()

One matched or unmatched flow in an activity comparison.

FieldTypeDefault
keystr
leftfloat | None
rightfloat | None
unitstr | None

One bucket inside an AggregateResult.

FieldTypeDefault
keystr
quantityfloat
countint
unitstr | NoneNone
sharefloat | NoneNone

Result of a Client.aggregate() call.

filtered_total is the sum across all items matching the filters (the top-level number). groups is the per-bucket breakdown when group_by was set; empty otherwise.

FieldTypeDefault
scopestr
filtered_totalfloat
filtered_unitstr | None
filtered_countint
groupslist[AggregateGroup]list()

An exchange with the environment (resource extraction or emission).

FieldTypeDefault
flow_namestr
flow_categorystr
amountfloat
unitstr
is_inputbool
commentstr | NoneNone
is_biosphereboolTrue

Filter a supply-chain/consumers query by a classification (system, value, mode).

Matches one classification system entry (e.g. (“Category”, “Agricultural\Food”, “exact”)). Mode is “exact” (case-insensitive equality) or “contains” (substring). Multiple filters are AND-combined by the server.

FieldTypeDefault
systemstr
valuestr
modestr’contains’

Activity that consumes a given supplier, with BFS depth.

FieldTypeDefault
process_idstr
namestr
locationstr
productstr
product_amountfloat
product_unitstr
depthint
classificationsdict[str, str]dict()

Reverse supply chain (/consumers) — paginated consumer list plus optional edge set. Mirrors :class:SupplyChain so callers have a uniform {entries, edges} shape in both traversal directions. edges is populated only when include_edges=True.

FieldTypeDefault
consumerslist[ConsumerResult]
totalint
offsetint
limitint
has_morebool
search_time_msfloat
edgeslist[SupplyChainEdge]list()

One entry of :meth:Client.list_databases.

depends_on names the databases this one links against for cross-DB flow resolution — mirrors the dependsOn list surfaced by the relink endpoint. Derived from the engine’s declared topology, not runtime state.

FieldTypeDefault
namestr
display_namestr
statusstr
pathstr
load_at_startupboolFalse
is_uploadedboolFalse
activity_countint0
descriptionstr | NoneNone
formatstr | NoneNone
depends_onlist[str]list()

Top contributing elementary flow for an impact category.

Emitted inside LCIAResult.top_contributors.

FieldTypeDefault
flow_namestr
contributionfloat
share_pctfloat
flow_idstr
categorystr
cf_valuefloat0.0
compartmentstr | NoneNone

Result of :func:download.

FieldTypeDefault
binaryPath
data_dirPath
versionstr
data_versionstr

Batch LCIA: every impact category in a method collection, for one activity.

Returned by :meth:Client.get_impacts_batch. Carries the per-method impact results plus any formula-based scoring sets configured in the engine TOML (PEF, ECS, or any named set).

scoring_indicators gives the per-variable normalized-weighted breakdown of each scoring set — already multiplied by the set’s displayMultiplier and expressed in its display unit (see :class:ScoringIndicator). Lets callers render per-indicator charts alongside the aggregate scoring_results.

FieldTypeDefault
resultslist[LCIAResult]
single_scorefloat | NoneNone
single_score_unitstr | NoneNone
norm_weight_set_namestr | NoneNone
available_nw_setslist[str]list()
scoring_resultsdict[str, dict[str, float]]dict()
scoring_unitsdict[str, str]dict()
scoring_indicatorsdict[str, dict[str, ScoringIndicator]]dict()

LCIA score for one impact category on one activity.

Returned directly by :meth:Client.get_impacts, and nested inside :class:LCIABatchResult.results (one entry per impact category).

FieldTypeDefault
method_idstr
method_namestr
categorystr
damage_categorystr
scorefloat
unitstr
mapped_flowsint
functional_unitstr
normalized_scorefloat | NoneNone
weighted_scorefloat | NoneNone
top_contributorslist[FlowContribution]list()

Shortest upstream path from a root process to a matching activity.

FieldTypeDefault
pathlist[PathStep]
path_lengthint
total_ratiofloat

One step in the supply chain path returned by get_path_to.

Note: the /path endpoint emits snake_case JSON directly (built via aeson’s object [...] rather than generic ToJSON), so it bypasses the engine’s stripLowerPrefix transform.

FieldTypeDefault
process_idstr
namestr
locationstr
unitstr
cumulative_quantityfloat
scaling_factorfloat
local_step_ratiofloat | NoneNone

One per-variable entry inside LCIABatchResult.scoring_indicators.

value is pre-multiplied by the scoring set’s displayMultiplier (configured in the scoring TOML) and expressed in the set’s display unit. category names the impact category the variable was resolved from.

FieldTypeDefault
categorystr
valuefloat
FieldTypeDefault
rootActivity
total_activitiesint
filtered_activitiesint
entrieslist[SupplyChainEntry]list()
edgeslist[SupplyChainEdge]list()

from/to are Python keywords, so they’re stored under from_id/to_id.

FieldTypeDefault
from_idstr
to_idstr
amountfloat
FieldTypeDefault
process_idstr
namestr
locationstr
quantityfloat
unitstr
scaling_factorfloat
classificationsdict[str, str]dict()

An exchange with another activity (input or output of an intermediate product).

Built from an ExchangeWithUnit envelope: outer fields like flowName/unitName live next to an inner exchange object (the discriminated Exchange sum).

FieldTypeDefault
flow_namestr
flow_categorystr
amountfloat
unitstr
is_inputbool
is_referencebool
target_activitystr | None
target_locationstr | None
target_process_idstr | None
commentstr | NoneNone
is_biosphereboolFalse

compare_activities(client: Client, pid_left: str, pid_right: str, *, scope: str = 'direct', group_by: str = 'flow_id', is_input: bool | None = True, **aggregate_kwargs) -> ActivityDiff

Section titled “compare_activities(client: Client, pid_left: str, pid_right: str, *, scope: str = 'direct', group_by: str = 'flow_id', is_input: bool | None = True, **aggregate_kwargs) -> ActivityDiff”

Diff two activities by flow_id (default) at the requested scope.

Returns three lists:

  • matched: flows present in both activities (with left, right, delta).
  • left_only: flows present only in the left activity.
  • right_only: flows present only in the right activity.

Default is_input=True restricts the comparison to inputs, which is the common case for “what does this variant consume differently?”. Pass is_input=None to include outputs as well.

download(version: Optional[str] = None, repo: str = 'ccomb/volca', *, force: bool = False) -> Installed

Section titled “download(version: Optional[str] = None, repo: str = 'ccomb/volca', *, force: bool = False) -> Installed”

Download the volca binary + data bundle for the current platform.

Idempotent: if both artefacts are already extracted under the install root and force=False, returns immediately without network.

Args: version: GH Release tag (v0.7.0); None resolves the latest. repo: GitHub repo slug. Default ccomb/volca. force: Re-download even if the install root looks complete.

Returns: :class:Installed with the resolved paths and versions.

Type alias: Union[TechnosphereExchange, BiosphereExchange].