antelope.models package

Submodules

antelope.models.auth module

exception antelope.models.auth.NoAuthorization

Bases: Exception

An error to signal that no authorization is found

class antelope.models.auth.AuthModel

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.auth.AuthorizationGrant(*, user: str, origin: str, issuer: str = None, access: str, values: bool = False, update: bool = False)

Bases: AuthModel

This class stands alone as a list of authorizations granted to users. There is no requirement that the users exist in a database

One natural way to use this is to grant users JWTs as bearer tokensusing an oauth2 system- the bearer token would be created + signed by an auth server, and include the grants embedded in the token. In the future, this token could list specific entities the user has a right to access (e.g. per ecoinvent). The problem is that entity UUIDs are 36 characters long- this would become cumbersome if the user had access to a great many entities (e.g. this doesn’t fly for user foreground grants– but– )

The way we are planning on doing this is embedding the token in the RESOURCE, so that the client simply passes it- in which case every resource would have its own token. AND the tokens would not be user- specific. Basically it becomes not the xdb’s job to check the user’s identity, only to validate the token and confirm that the token provides the requested access. (and log and meter the users’ requests).

Eventually we may want to get an auth db back in the loop and have xdb check to see whether a user can access a given entity / or enforce a quota.

user: str
origin: str
issuer: str
access: str
values: bool
update: bool
authorizes(origin)

grant authorizes origin if origin is of equal or greater specificity (origin and sub-origins) :param origin: :return:

serialize()

# old spec from blackbook draft grants are serialized as ‘ ‘-delimited access specifications of the form:

origin.dot.separated[:interface[,f]*]

interfaces only need to be three-letter, according to:

bas-ic ind-ex exc-hange bac-kground qua-ntity for-eground

,-separated Flags follow:

v - values w - writable m - metered

The origin ‘qdb’ is special and refers to antelope cloud qdb and is short for “qdb:basic:quantity,v” (lots of DNS TXT record vibes here)

An example of a verbose, precise specification would be:

antelope: “ecoinvent.3.6.cut-off:basic:index:exchange,v:background,v,m qdb:quantity,v”

An example of a terse, broad specification would be:

antelope: “ecoinvent:bas:ind:exc,v:bac,v,m qdb”

The __str__ function for Grant objects the components between the colons in canonical (2-letter) form

property display
classmethod from_jwt(jwt: JwtGrant) List

returns a list of grants NOTE: This does not currently deal with quota flags :param jwt: :return:

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Module contents

Pydantic Model definitions for data exchange.

TODO: decide on a consistent structure / format for entity identification. Currently:

  • EntityRef and subclasses use entity_id

  • FlowSpec uses external_ref

  • ExteriorFlow uses flow

  • Exchange and subclasses use <type> e.g. process and flow.external_ref (exchange.flow is FlowSpec)

  • Quantity-related subclasses use <mod>_<type> e.g. ref_quantity, query_quantity

= This tells me that FragmentFlows should use ‘fragment’ and ‘parent_fragment’

class antelope.models.ResponseModel

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.OriginMeta(*, origin: str, is_lcia_engine: bool, interfaces: List[str])

Bases: ResponseModel

origin: str
is_lcia_engine: bool
interfaces: List[str]

based on auth token, what interfaces can the user access? read: List(str) write: List(str) values: List(str)

Type:

# report the authorization status of the query

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.OriginCount(*, origin: str, count: dict)

Bases: ResponseModel

origin: str
count: dict
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.EntityRef(*, origin: str, entity_id: str, entity_type: str | None)

Bases: ResponseModel

origin: str
entity_id: str
entity_type: str | None
property external_ref
classmethod from_entity(entity)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.Entity(*, origin: str, entity_id: str, entity_type: str, properties: Dict)

Bases: EntityRef

entity_type: str
properties: Dict
signature_fields()
classmethod from_entity(entity)
classmethod from_flow(flow)
classmethod from_json(j)
serialize()

this simulates LcEntity.serialize() :return:

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.FlowEntity(*, origin: str, entity_id: str, entity_type: str, properties: Dict, context: List[str], locale: str)

Bases: Entity

Open questions: should context and locale be properties? or attributes? should referenceQuantity be an attribute?

context: List[str]
locale: str
classmethod from_flow(entity, **kwargs)
classmethod from_exchange(ex)

This is from an antelope Exchange or ExchangeRef :param ex: :return:

classmethod from_exchange_model(ex)

ExchangeModel formerly included a FlowSpec instead of a flow– but now it includes a FlowEntity :param ex: :return:

serialize()

this simulates LcEntity.serialize() :return:

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.Context(*, name: str, parent: str | None, sense: str | None, elementary: bool, subcontexts: List[str])

Bases: ResponseModel

name: str
parent: str | None
sense: str | None
elementary: bool
subcontexts: List[str]
classmethod from_context(cx)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.FlowSpec(*, origin: str | None = None, external_ref: str | None = None, flowable: str, quantity_ref: str, context: List[str], locale: str)

Bases: ResponseModel

Non-context exchange terminations do not get included in flow specifications, because they are part of the model and not the flow

origin: str | None
external_ref: str | None
flowable: str
quantity_ref: str
context: List[str]
locale: str
classmethod from_flow(flow)
classmethod from_termination(term)
classmethod from_exchange(x, locale=None)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.DirectedFlow(*, flow: FlowSpec, direction: str)

Bases: ResponseModel

flow: FlowSpec
direction: str
property origin
property external_ref
property quantity_ref
property name
classmethod from_observed(obj)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.ExteriorFlow(*, flow: FlowSpec, direction: str, context: List[str])

Bases: DirectedFlow

An ExteriorFlow is essentially a row in the LCI Environment B matrix. It consists of a directed flow, enhanced with a context. Now I know a flow already has a context, but (a) context is not required for a flow and (b) flows can be terminated to contexts other than their ‘default’

context: List[str]
property termination
classmethod from_background(flow, direction, context)
classmethod from_exterior(obj)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.Exchange(*, origin: str, process: str, flow: FlowEntity, direction: str, termination: str | None, context: List[str] | None, type: str, comment: str | None)

Bases: ResponseModel

Do we need to add locale?? no it’s in the flow entity

origin: str
process: str
flow: FlowEntity
direction: str
termination: str | None
context: List[str] | None
type: str
comment: str | None
classmethod from_exchange(x, **kwargs)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.ReferenceExchange(*, origin: str, process: str, flow: FlowEntity, direction: str, termination: None, context: List[str] | None, type: str, comment: str | None, is_reference: bool = True)

Bases: Exchange

is_reference: bool
termination: None
classmethod from_exchange(x, **kwargs)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.ReferenceValue(*, origin: str, process: str, flow: FlowEntity, direction: str, termination: None, context: List[str] | None, type: str, comment: str | None, is_reference: bool = True, value: float = 0.0)

Bases: ReferenceExchange

value: float
classmethod from_rx(x)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.ExchangeValues(*, origin: str, process: str, flow: FlowEntity, direction: str, termination: str | None, context: List[str] | None, type: str, comment: str | None, values: Dict, uncertainty: Dict | None = None)

Bases: Exchange

dict mapping reference flows to allocated value This should really be called ExchangeValues- in fact the method is already called exchangeValues!

values: Dict
uncertainty: Dict | None
classmethod from_ev(x)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.UnallocatedExchange(*, origin: str, process: str, flow: FlowEntity, direction: str, termination: str | None, context: List[str] | None, type: str, comment: str | None, is_reference: bool = False, value: float, uncertainty: Dict | None = None)

Bases: Exchange

is_reference: bool
value: float
uncertainty: Dict | None
classmethod from_inv(x)

This works for ExchangeRefs as well as exchanges :param x: :return:

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.AllocatedExchange(*, origin: str, process: str, flow: FlowEntity, direction: str, termination: str | None, context: List[str] | None, type: str, comment: str | None, ref_flow: str, value: float, uncertainty: Dict | None = None)

Bases: Exchange

ref_flow: str
value: float
uncertainty: Dict | None
property is_reference
classmethod from_inv(x, ref_flow: str)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

antelope.models.generate_pydantic_exchanges(xs, type=None)
Parameters:
  • xs – iterable of exchanges

  • type – [None] whether to filter the exchanges by type. could be one of None, ‘reference’, ‘self’, ‘node’,

‘context’, ‘cutoff’ :return:

antelope.models.generate_pydantic_inventory(xs, mode=None, values=False, ref_flow=None)

Not currently used

Parameters:
  • xs – iterable of exchanges

  • mode – [None] whether to filter the exchanges by type. could be one of: - None: generate all exchanges - ‘interior’ - ‘exterior’ - ‘cutoff’

  • values – (bool) [False] whether to include exchange values.

  • ref_flow – (ignored if values=False) the reference flow with which the exchange value was computed. If None, this implies the exchange reports un-allocated exchange values

Returns:

class antelope.models.Characterization(*, origin: str, flowable: str, ref_quantity: str, ref_unit: str | None = None, query_quantity: str, query_unit: str | None = None, context: List[str], value: Dict)

Bases: ResponseModel

origin: str
flowable: str
ref_quantity: str
ref_unit: str | None
query_quantity: str
query_unit: str | None
context: List[str]
value: Dict
classmethod from_cf(cf)
classmethod from_cfs(cfs)
classmethod null(cf)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.Normalizations(*, origin: str, quantity: str, norm: Dict, weight: Dict)

Bases: ResponseModel

This is written to replicate the normalisation data stored per-method in OpenLCA JSON-LD format

origin: str
quantity: str
norm: Dict
weight: Dict
classmethod from_q(q)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.QuantityConversion(*, origin: str, flowable: str, ref_quantity: str, query_quantity: str, context: List[str], locale: str, value: float)

Bases: ResponseModel

Technically, a quantity conversion can include chained QR Results, but we are flattening it (for now)

origin: str
flowable: str
ref_quantity: str
query_quantity: str
context: List[str]
locale: str
value: float
classmethod from_qrresult(qrr)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.FlowFactors(*, origin: str, external_ref: str, context: List[str], factors: List[QuantityConversion])

Bases: ResponseModel

Client POSTs a list of FlowSpecs; server returns a list of characterizations that match the specs, grouped by flow external_ref (so as to be cached in the flow’s chars_seen).

The challenge here is with contexts: in order for lookup_cf to find the CF, it needs to be cached with a local context; but in order for the results to be portable/reproducible, the QR results should report the canonical contexts. So, we add a context field where we reproduce the posted context.

origin: str
external_ref: str
context: List[str]
factors: List[QuantityConversion]
add_qr_result(qrr)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.AggregatedLciaScore(*, origin: str, entity_id: str, component: str, result: float)

Bases: ResponseModel

origin: str
entity_id: str
component: str
result: float
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.SummaryLciaScore(*, origin: str, entity_id: str, component: str, result: float, node_weight: float | None, unit_score: float | None)

Bases: AggregatedLciaScore

node_weight: float | None
unit_score: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.LciaDetail(*, exchange: ExteriorFlow | None, factor: QuantityConversion, result: float)

Bases: ResponseModel

exchange: ExteriorFlow | None
factor: QuantityConversion
result: float
classmethod from_detailed_lcia_result(d)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.DisaggregatedLciaScore(*, origin: str, entity_id: str, component: str, result: float, details: List[LciaDetail] = [])

Bases: AggregatedLciaScore

details: List[LciaDetail]

@classmethod def from_component(cls, obj, c):

if hasattr(c.entity, ‘name’):

component = c.entity.name

else:

component = str(c.entity)

if hasattr(c.entity, ‘external_ref’):

origin = c.entity.origin entity_id = c.entity.external_ref

else:

origin = obj.origin entity_id = obj.external_ref

return cls(origin=origin, entity_id=entity_id, component=component, result=c.cumulative_result,
details=[LciaDetail.from_detailed_lcia_result(d) for d in

sorted(c.details(), key=lambda x: x.result, reverse=True)])

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class antelope.models.LciaResult(*, scenario: str | None, object: str, quantity: Entity, scale: float, total: float, components: List[DisaggregatedLciaScore] = [], summaries: List[SummaryLciaScore] = [])

Bases: ResponseModel

Note that a core LciaResult can contain either (detailed) components or summaries, but not both.

scenario: str | None
object: str
quantity: Entity
scale: float
total: float
components: List[DisaggregatedLciaScore]
summaries: List[SummaryLciaScore]
classmethod from_lcia_result(obj, res)
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod summary(obj, res)
classmethod detailed(obj, res)