app.schemas package

Submodules

app.schemas.connector module

Connector schemas.

This module defines the Pydantic schemas used to represent and validate connector-related data in API responses and updates.

Schemas:

ConnectorResponse: Returned by the API when retrieving connector details.
ConnectorUpdate: Used when updating an existing connector.

These schemas ensure consistency and proper validation for data exchanged between the frontend and the backend.

class app.schemas.connector.ConnectorResponse(**data)

Bases: BaseModel

Represents a connector object returned by the API.

This schema defines the structure of connector data retrieved from the database or exposed through the REST API.

Example

>>> connector = ConnectorResponse(
...     id="edc-provider-01",
...     name="EDC Provider",
...     type="provider",
...     state="running",
...     mode="managed",
...     ports=PortConfig(http=8181, management=9999, protocol=8282, control=9191, public=8182, version=1)
... )
>>> print(connector.name)
EDC Provider

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

id: str: Unique identifier of the connector.

name: str: Human-readable name of the connector.

description: Optional[str]: Optional description providing additional connector details.

type: Literal['provider', 'consumer']: Role of the connector in data exchange (provider or consumer).

ports: Optional[PortConfig]: Port configuration details for the connector.

state: Literal['running', 'stopped']: Operational state of the connector (running or stopped).

mode: Literal['managed', 'remote']: Indicates whether the connector runs locally (managed) or remotely (remote).

endpoints_url: Optional[Endpoints]: Management and protocol endpoint URLs associated with the connector.

api_key: Optional[str]: Authentication key for the connector, if applicable.

domain: Optional[str]

Public domain name of the connector.

Specifies the external hostname where the connector is accessible (e.g., edc-provider.mycompany.com). This field is required when the connector is deployed on a remote server and must be reachable by other EDC connectors.

For local environments or isolated testing, this field can be left empty or set to localhost.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.connector.ConnectorUpdate(**data)

Bases: BaseModel

Represents the schema used to update existing connector configurations.

All fields are optional to allow partial updates.

Example

>>> update = ConnectorUpdate(
...     name="Updated Provider",
...     state="stopped",
...     mode="unmanaged"
... )
>>> print(update.state)
stopped

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: Optional[str]: Updated connector name.

description: Optional[str]: Updated description for the connector.

type: Optional[Literal['provider', 'consumer']]: Updated connector type (provider or consumer).

ports: Optional[PortConfig]: Updated port configuration.

api_key: Optional[str]: Updated authentication key, if applicable.

state: Optional[Literal['running', 'stopped']]: Updated operational state (running or stopped).

mode: Optional[Literal['managed', 'unmanaged']]: Updated deployment mode (managed or unmanaged).

endpoints_url: Optional[Endpoints]: Updated management and protocol endpoint URLs.

domain: Optional[str]

Public domain name of the connector.

Specifies the external hostname where the connector is accessible (e.g., edc-provider.mycompany.com). This field is required when the connector is deployed on a remote server and must be reachable by other EDC connectors.

For local environments or isolated testing, this field can be left empty or set to localhost.

app.schemas.transfer module

Transfer schemas.

This module defines the Pydantic schemas used to represent and validate data exchange operations between EDC connectors. These schemas are used for both API requests and responses related to catalog retrieval, contract negotiation, transfer initiation, and status checks.

Schemas:

TransferResponse: Full transfer record returned by the API.
RequestCatalog: Structure for requesting a provider’s data catalog.
NegotitateContract: Structure for initiating contract negotiation.
ContractAgreement: Structure for retrieving an established agreement.
StartTransfer: Structure for starting a data transfer.
CheckTransfer: Structure for checking transfer status.

class app.schemas.transfer.TransferResponse(**data)

Bases: BaseModel

Represents a complete data transfer record between two EDC connectors.

This schema captures all metadata related to a transfer, including contract identifiers, transfer process information, and flow type.

Example

>>> transfer = TransferResponse(
...     id="transfer-001",
...     consumer="edc-consumer-01",
...     provider="edc-provider-01",
...     asset="asset-001",
...     has_policy_id="policy-001",
...     negotiate_contract_id="neg-001",
...     contract_agreement_id="agreement-001",
...     transfer_process_id="process-001",
...     transfer_flow="push"
... )
>>> print(transfer.transfer_flow)
push

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

id: str: Unique identifier of the transfer record.

consumer: str: Identifier of the consumer EDC connector.

provider: str: Identifier of the provider EDC connector.

asset: str: Identifier of the asset involved in the transfer.

has_policy_id: str: Identifier of the policy governing this transfer.

negotiate_contract_id: str: Identifier of the contract negotiation process.

contract_agreement_id: str: Identifier of the established contract agreement.

transfer_process_id: str: Identifier of the transfer process in the EDC connector.

transfer_flow: Literal['push', 'pull']

push or pull.

Type:: Type of transfer flow

authorization: Optional[str]: Optional authorization token used for secured transfers.

endpoint: Optional[str]: Optional endpoint URL used for accessing or delivering data.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.transfer.RequestCatalog(**data)

Bases: BaseModel

Represents a catalog request between two connectors.

Used to retrieve the list of assets available for negotiation from a provider connector.

Example

>>> request = RequestCatalog(
...     consumer="edc-consumer-01",
...     provider="edc-provider-01"
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

consumer: str: Identifier of the consumer connector requesting the catalog.

provider: str: Identifier of the provider connector offering assets.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class app.schemas.transfer.NegotitateContract(**data)

Bases: BaseModel

Represents a contract negotiation request between two connectors.

Used to initiate a negotiation for a specific asset based on a given contract offer ID.

Example

>>> negotiation = NegotitateContract(
...     consumer="edc-consumer-01",
...     provider="edc-provider-01",
...     contract_offer_id="offer-001",
...     asset="asset-001"
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

consumer: str: Identifier of the consumer connector.

provider: str: Identifier of the provider connector.

contract_offer_id: str: Identifier of the offered contract to be negotiated.

asset: str: Identifier of the asset associated with this contract offer.

class app.schemas.transfer.ContractAgreement(**data)

Bases: BaseModel

Represents a request to retrieve a finalized contract agreement.

Example

>>> agreement = ContractAgreement(
...     consumer="edc-consumer-01",
...     id_contract_negotiation="neg-001"
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

consumer: str: Identifier of the consumer connector.

id_contract_negotiation: str: Identifier of the previously negotiated contract.

class app.schemas.transfer.StartTransfer(**data)

Bases: BaseModel

Represents the information required to start a transfer process.

Example

>>> start = StartTransfer(
...     consumer="edc-consumer-01",
...     provider="edc-provider-01",
...     contract_agreement_id="agreement-001"
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

consumer: str: Identifier of the consumer connector initiating the transfer.

provider: str: Identifier of the provider connector offering the data.

contract_agreement_id: str: Identifier of the established contract agreement.

class app.schemas.transfer.CheckTransfer(**data)

Bases: BaseModel

Represents a request to check the status of a transfer process.

Example

>>> check = CheckTransfer(
...     consumer="edc-consumer-01",
...     transfer_process_id="process-001"
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

consumer: str: Identifier of the consumer connector.

transfer_process_id: str: Identifier of the transfer process to check.

app.schemas package

Submodules

app.schemas.connector module

app.schemas.transfer module

Module contents