fluvius-energy-api 0.2.2

Python wrapper for the Fluvius Energy API

Fluvius Energy API

Python wrapper for the Fluvius Energy API v3 (ESCO - Energy Service Company API).

Installation

pip install fluvius-energy-api

Or with uv:

uv add fluvius-energy-api

Configuration

The client requires OAuth2 credentials for Azure AD authentication. Two authentication methods are supported:

• Certificate-based (production): Uses certificate thumbprint + private key
• Client secret (sandbox): Uses a simple client secret

Common Environment Variables

These are shared between production and sandbox:

Variable	Description
FLUVIUS_SUBSCRIPTION_KEY	Azure API Management subscription key
FLUVIUS_CLIENT_ID	Azure AD application (client) ID
FLUVIUS_TENANT_ID	Azure AD tenant ID
FLUVIUS_SCOPE	OAuth2 scope (e.g., api://your-app-id/.default)

Production (Certificate-based auth)

Variable	Description
FLUVIUS_CERTIFICATE_THUMBPRINT	Certificate thumbprint (hex format)
FLUVIUS_PRIVATE_KEY	RSA private key in PEM format
FLUVIUS_PRIVATE_KEY_PATH	Alternative: path to private key file

Sandbox (Client secret auth)

Variable	Description
FLUVIUS_SANDBOX_CLIENT_SECRET	Client secret for sandbox

Note: For sandbox, common values (subscription_key, client_id, tenant_id, scope, data_access_contract_number) automatically fall back to FLUVIUS_* if not set with FLUVIUS_SANDBOX_* prefix.

Data Access Contract

Variable	Description
FLUVIUS_DATA_ACCESS_CONTRACT_NUMBER	Data access contract number (required, typically constant per service provider)

Private Key Formats

The private key can be provided in three ways:

1. Inline PEM: Set FLUVIUS_PRIVATE_KEY with the full PEM content
2. Base64-encoded: Set FLUVIUS_PRIVATE_KEY with the PEM content encoded as base64
3. File path: Set FLUVIUS_PRIVATE_KEY_PATH to point to a .pem file

Usage

Production (Certificate auth)

from fluvius_energy_api import FluviusEnergyClient, Environment

# Credentials loaded from FLUVIUS_* environment variables
with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    response = client.get_mandates()
    for mandate in response.data.mandates:
        print(f"EAN: {mandate.ean}, Status: {mandate.status}")

Sandbox (Client secret auth)

from fluvius_energy_api import FluviusEnergyClient, FluviusCredentials, Environment

# Load credentials - falls back to FLUVIUS_* for common values,
# uses FLUVIUS_SANDBOX_CLIENT_SECRET for auth
sandbox_credentials = FluviusCredentials.from_env(prefix="FLUVIUS_SANDBOX")

with FluviusEnergyClient(credentials=sandbox_credentials, environment=Environment.SANDBOX) as client:
    response = client.get_mandates()
    print(f"Found {len(response.data.mandates)} mandates in sandbox")

Explicit Credentials

from fluvius_energy_api import FluviusEnergyClient, FluviusCredentials, Environment

# Production with certificate
prod_credentials = FluviusCredentials(
    subscription_key="your-subscription-key",
    client_id="your-client-id",
    tenant_id="your-tenant-id",
    scope="api://your-scope/.default",
    certificate_thumbprint="YOUR_CERTIFICATE_THUMBPRINT",
    private_key="-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----",
)

# Sandbox with client secret
sandbox_credentials = FluviusCredentials(
    subscription_key="your-subscription-key",
    client_id="your-client-id",
    tenant_id="your-tenant-id",
    scope="api://your-scope/.default",
    client_secret="your-client-secret",
)

with FluviusEnergyClient(credentials=prod_credentials, environment=Environment.PRODUCTION) as client:
    mandates = client.get_mandates()

Create Client Session

Create a session for end-users to grant data access:

from fluvius_energy_api import (
    FluviusEnergyClient,
    ClientSessionDataService,
    DataServiceType,
    Environment,
)

with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    response = client.create_client_session(
        data_access_contract_number="3599900040",
        reference_number="my-unique-reference",
        flow="B2B",
        data_services=[
            ClientSessionDataService(
                data_service_type=DataServiceType.VH_DAG,
                data_period_from="2024-01-01T00:00:00Z",
            ),
            ClientSessionDataService(
                data_service_type=DataServiceType.VH_KWARTIER_UUR,
                data_period_from="2024-01-01T00:00:00Z",
            ),
        ],
        number_of_eans=5,
    )

    if response.data.status == "success":
        # Build the consent URL for end-users
        consent_url = f"https://mijn.fluvius.be/verbruik/dienstverlener?id={response.data.short_url_identifier}"
        print(f"Consent URL: {consent_url}")
        print(f"Valid until: {response.data.valid_to}")

Get Mandates

Retrieve mandates with optional filters:

from fluvius_energy_api import (
    FluviusEnergyClient,
    EnergyType,
    MandateStatus,
    DataServiceType,
    Environment,
)

with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    # Get all mandates
    response = client.get_mandates()

    # Get mandates with filters
    response = client.get_mandates(
        ean="541448800000004312",
        energy_type=EnergyType.ELECTRICITY,
        status=MandateStatus.APPROVED,
        data_service_types=[DataServiceType.VH_DAG, DataServiceType.VH_KWARTIER_UUR],
    )

    for mandate in response.data.mandates:
        print(f"EAN: {mandate.ean}")
        print(f"Status: {mandate.status}")
        print(f"Energy type: {mandate.energy_type}")
        print(f"Data service: {mandate.data_service_type}")

Get Energy Data

Retrieve energy consumption/production data:

from datetime import datetime
from fluvius_energy_api import FluviusEnergyClient, PeriodType, Environment

with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
    response = client.get_energy(
        ean="541448800000004312",
        period_type=PeriodType.READ_TIME,
        granularity="daily",
        from_date=datetime(2024, 1, 1),
        to_date=datetime(2024, 6, 1),
    )

    headpoint = response.data.headpoint
    print(f"EAN: {headpoint.ean}")
    print(f"Energy type: {headpoint.energy_type}")

Postman Collection

A hand-crafted Postman collection is included in the postman/ directory covering authentication, client sessions, mandates, and energy data endpoints.

To generate the environment files from your .env:

python postman/generate_postman_env.py

This creates two files (gitignored):

• postman/fluvius-sandbox.postman_environment.json -- client-secret auth
• postman/fluvius-production.postman_environment.json -- certificate / JWT auth

Import the collection (postman/fluvius-energy-api.postman_collection.json) and the appropriate environment into Postman. The collection's pre-request script acquires tokens automatically.

The production client assertion is valid for 1 hour. Re-run the script for a fresh one -- the JWT is printed to stdout so you can copy-paste it directly into your already-imported Postman environment without re-importing the file.

Environments

The client supports two environments:

Environment	Base URL	Auth Method
Environment.PRODUCTION	https://apihub.fluvius.be/esco-live/v3	Certificate
Environment.SANDBOX	https://apihub.fluvius.be/esco-sbx/v3	Client secret

from fluvius_energy_api import FluviusEnergyClient, Environment

# Production
client = FluviusEnergyClient(environment=Environment.PRODUCTION)

# Sandbox
client = FluviusEnergyClient(environment=Environment.SANDBOX)

Error Handling

The client raises specific exceptions for different error scenarios:

from fluvius_energy_api import (
    FluviusEnergyClient,
    FluviusAPIError,
    AuthenticationError,
    ForbiddenError,
    NotFoundError,
    ValidationError,
    ServerError,
    ServiceUnavailableError,
    ConfigurationError,
    Environment,
)

try:
    with FluviusEnergyClient(environment=Environment.PRODUCTION) as client:
        response = client.get_mandates()
except ConfigurationError as e:
    print(f"Configuration error: {e}")
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except ForbiddenError as e:
    print(f"Access forbidden: {e}")
except NotFoundError as e:
    print(f"Resource not found: {e}")
except ValidationError as e:
    print(f"Validation error: {e}")
    if e.validation_errors:
        for error in e.validation_errors:
            print(f"  - {error}")
except ServerError as e:
    print(f"Server error: {e}")
except ServiceUnavailableError as e:
    print(f"Service unavailable: {e}")
except FluviusAPIError as e:
    print(f"API error: {e}")

Data Service Types

Available data service types:

Type	Description
VH_DAG	Daily consumption history
VH_KWARTIER_UUR	Quarter-hourly/hourly consumption history
VH_ONBEPAALD	Indefinite consumption history
IG	Installation data

License

GNU AFFERO GENERAL PUBLIC LICENSE