Python

Starting point για services, workers και scheduled jobs που καλούν το Estia API.

Το κύριο που πρέπει να λυθεί από νωρίς: token reuse. Νέο token σε κάθε request δουλεύει σε demo αλλά σε production φέρνει θόρυβο και άσκοπη πολυπλοκότητα.

Full example με 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()

Παράδειγμα χρήσης

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

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

Async εκδοχή με 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()