Outbound RESTful Invoker Service

Overview

The REST Invoker Basic System streamlines the integration with external API systems by automating data collection and processing within the energy market ecosystem. This solution enables organisations to:

• Automatically fetch critical market data such as reconciliation prices, settlement prices, and balance information
• Securely manage API connections with built-in authentication handling
• Standardise data collection processes across different market systems
• Reduce development time through simple, reusable integration patterns
• Ensure reliable data storage through automated Google Cloud Storage integration

RIBS significantly reduces the complexity of market data integration while ensuring secure and reliable data collection. By automating routine data gathering tasks, it allows organisations to focus on data analysis rather than data collection, leading to improved operational efficiency and faster market insights.

REST to GCS

Sends a request to a specified external API system and publishes the response to a custom Pub/Sub topic. This method facilitates integration with external APIs defined in the ewx-ribs configuration.

Quick Start

Here's a simple example of fetching reconciliation prices:

    from datetime import datetime
    from energyworx.domain import RuleResult
    from energyworx.rules.base_rule import AbstractRule
    from ribs import Ribs
    class ReconciliationPriceFetcher(AbstractRule):
        def apply(self, **kwargs) -> RuleResult:
            date_from = datetime(2024, 1, 1).isoformat()
            date_to = datetime(2024, 2, 1).isoformat()
            return Ribs.rest_to_gcs(
                self,
                ribs_system_id='api-dev-tennet-eu',
                resource_path='v1/reconciliation-prices',
                params={
                    "date_from": date_from,
                    "date_to": date_to
                },
                # accept="application/xml"  # uncomment to switch to XML
                # accept="text/csv"  # uncomment to switch to CSV
            )

Authentication is automatically handled****by ewx-ribs - you cannot supply any authentication headers or tokens.

Parameters

ribs_system_id: str

The system identifier for the external API as defined in targets.yaml.

Supported TenneT systems:

• api-dev-tennet-eu: TenneT development environment
• api-tennet-eu: TenneT production environment

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/reconciliation-prices'
    )

resource_path: str

The endpoint path of the external API, excluding the base URL.

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/reconciliation-prices'
    )

data: bytes

Optional data to send in the request body. Defaults to b"x" as the technical stack requires non-empty messages.

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/balance-delta',
        data=json.dumps({"key": "value"}).encode()
    )

m_id: str | None

Optional trace ID for the job. If not provided, a random UUID is generated.

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/settlement-prices',
        m_id='custom-trace-id-123'
    )

headers: dict[str, str] | None

Optional headers to include in the request.
Note: Authentication headers are automatically handled by ewx-ribs.

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/reconciliation-prices',
        headers={
            'Custom-Header': 'value'
        }
    )

params: dict[str, str] | None

Optional query parameters to include in the request URL.

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/reconciliation-prices',
        params={
            'date_from': '2024-01-01T00:00:00',
            'date_to': '2024-02-01T00:00:00'
        }
    )

content_type: str

The Content-Type header for the request. Defaults to 'application/json'.

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/reconciliation-prices',
        content_type='application/x-www-form-urlencoded'
    )

accept: str

The Accept header for the request. Defaults to application/json.

This can only be one of:

• application/xml
• application/json
• text/csv

Example:

    Ribs.rest_to_gcs(
        self,
        ribs_system_id='api-dev-tennet-eu',
        resource_path='v1/reconciliation-prices',
        accept='application/xml'
    )

payload_type: str
Specifies the file or payload type, determining which market adapter will ingest the message. This parameter defines the type of payload used internally and can influence processing logic.
Example:

    Ribs.rest_to_gcs(
       self,
       ribs_system_id='api-dev-tennet-eu',
       resource_path='v1/reconciliation-prices',
       payload_type="application/xml",
       ...
    )

tags: list[str] | None
Optional tags to attach to the file in the file manager. Tags serve as metadata to help identify or categorize the file and are extended internally with additional information (e.g., creation timestamp, system ID, resource path).
Example:

    Ribs.rest_to_gcs(
       self,
       ribs_system_id='api-dev-tennet-eu',
       resource_path='v1/reconciliation-prices',
       payload_type="application/json",
       tags=["example-tag", "metadata-tag"],
       ...
    )

Complete Examples

Fetch Reconciliation Prices

    from datetime import datetime
    from energyworx.domain import RuleResult
    from energyworx.rules.base_rule import AbstractRule
    from ribs import Ribs
    class ReconciliationPriceFetcher(AbstractRule):
        def apply(self, **kwargs) -> RuleResult:
            # Note: Maximum range is 7 years for this endpoint
            date_from = datetime(2024, 1, 1).isoformat()
            date_to = datetime(2024, 2, 1).isoformat()
            return Ribs.rest_to_gcs(
                self,
                ribs_system_id='api-dev-tennet-eu',
                resource_path='v1/reconciliation-prices',
                params={
                    "date_from": date_from,
                    "date_to": date_to
                },           
            )

Fetch Settlement Prices

    from datetime import datetime, timedelta
    from energyworx.domain import RuleResult
    from energyworx.rules.base_rule import AbstractRule
    from ribs import Ribs
    class SettlementPriceFetcher(AbstractRule):
        def apply(self, **kwargs) -> RuleResult:
            # Note: Maximum range is 1 hour for this endpoint
            now = datetime.now()
            date_from = now
            date_to = now + timedelta(hours=1)
            return Ribs.rest_to_gcs(
                self,
                ribs_system_id='api-dev-tennet-eu',
                resource_path='v1/settlement-prices',
                params={
                    "date_from": date_from.isoformat(),
                    "date_to": date_to.isoformat()
                }
            )

Fetch Balance Delt

    from datetime import datetime, timedelta
    from energyworx.domain import RuleResult
    from energyworx.rules.base_rule import AbstractRule
    from ribs import Ribs
    class BalanceDeltaFetcher(AbstractRule):
        def apply(self, **kwargs) -> RuleResult:
            # Note: Maximum range is 1 day for this endpoint
            now = datetime.now()
            date_from = now
            date_to = now + timedelta(days=1)
            return Ribs.rest_to_gcs(
                self,
                ribs_system_id='api-dev-tennet-eu',
                resource_path='v1/balance-delta',
                params={
                    "date_from": date_from.isoformat(),
                    "date_to": date_to.isoformat()
                }
            )