Python

Starting point for services, workers, and scheduled jobs that call Estia API from Python.

The main thing to get right early: token reuse. A new token per request works in a demo, but in production it adds noise and unnecessary complexity.

Full example with requests

import os
import time
import threading
import requests
from typing import Any, Optional


class EstiaApiClient:
    def __init__(
        self,
        client_id: str,
        client_secret: str,
        api_base_url: str = "https://api.insurancegateway.gr",
        auth_url: str = "https://auth.insurancegateway.gr/realms/estia/protocol/openid-connect/token",
    ) -> None:
        self.client_id = client_id
        self.client_secret = client_secret
        self.api_base_url = api_base_url.rstrip("/")
        self.auth_url = auth_url
        self._token: Optional[str] = None
        self._token_expires_at: float = 0.0
        self._token_lock = threading.Lock()

    def _get_token(self) -> str:
        if self._token and time.time() < self._token_expires_at - 300:
            return self._token

        with self._token_lock:
            if self._token and time.time() < self._token_expires_at - 300:
                return self._token

            resp = requests.post(
                self.auth_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                },
                timeout=10,
            )
            resp.raise_for_status()
            body = resp.json()
            self._token = body["access_token"]
            self._token_expires_at = time.time() + body["expires_in"]
            return self._token

    def get(self, path: str) -> Any:
        return self._request("GET", path)

    def post(self, path: str, json_body: Any) -> Any:
        return self._request("POST", path, json=json_body)

    def _request(self, method: str, path: str, **kwargs: Any) -> Any:
        token = self._get_token()
        resp = requests.request(
            method,
            f"{self.api_base_url}{path}",
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
            **kwargs,
        )
        resp.raise_for_status()
        return resp.json()

Example usage

client = EstiaApiClient(
    client_id=os.environ["ESTIA_CLIENT_ID"],
    client_secret=os.environ["ESTIA_CLIENT_SECRET"],
)

brands = client.get("/intersalonica/auto/brands")

Async variant with httpx

import httpx

class AsyncEstiaApiClient:
    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self._token: Optional[str] = None
        self._token_expires_at: float = 0.0
        self._client = httpx.AsyncClient(base_url="https://api.insurancegateway.gr")

    async def _get_token(self) -> str:
        if self._token and time.time() < self._token_expires_at - 300:
            return self._token
        resp = await self._client.post(
            "https://auth.insurancegateway.gr/realms/estia/protocol/openid-connect/token",
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
            },
        )
        resp.raise_for_status()
        body = resp.json()
        self._token = body["access_token"]
        self._token_expires_at = time.time() + body["expires_in"]
        return self._token

    async def get(self, path: str) -> Any:
        token = await self._get_token()
        resp = await self._client.get(path, headers={"Authorization": f"Bearer {token}"})
        resp.raise_for_status()
        return resp.json()