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 ewx_public.rule_result import RuleResult
from ewx_public.flow_rule import FlowRule
from ribs import Ribs
class ReconciliationPriceFetcher(FlowRule):
    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.

Where does Ribs come from?

The Ribs helper used in these examples is provided by Energyworx as a library rule in your namespace — it is not a pip-installable package. If from ribs import Ribs is not available in your namespace, contact Energyworx to have the library rule installed.

Parameters

ribs_system_id: str

The system identifier for the external API. The available systems (base URL and authentication) are defined in the ewx-ribs service configuration (targets.yaml), which is maintained by Energyworx and shipped with the service — adding or changing an external system requires an Energyworx change and deployment, not a bucket configuration update.

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 ewx_public.rule_result import RuleResult
from ewx_public.flow_rule import FlowRule
from ribs import Ribs
class ReconciliationPriceFetcher(FlowRule):
    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 ewx_public.rule_result import RuleResult
from ewx_public.flow_rule import FlowRule
from ribs import Ribs
class SettlementPriceFetcher(FlowRule):
    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 Delta

from datetime import datetime, timedelta
from ewx_public.rule_result import RuleResult
from ewx_public.flow_rule import FlowRule
from ribs import Ribs
class BalanceDeltaFetcher(FlowRule):
    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()
            }
        )