session

session

Async HTTP session management for ETM API.

Classes

ETMResponse

Bases: BaseModel

Response object that works with both sync and async operations.

Attributes

content property

content

Get response content as bytes.

Functions

json

json()

Parse response body as JSON.

Source code in src/pyetm/clients/session.py

def json(self) -> Dict[str, Any]:  # type: ignore[override]
    """Parse response body as JSON."""
    if self._json_data is not None:
        return self._json_data

    import json

    return cast(Dict[str, Any], json.loads(self.text))

raise_for_status

raise_for_status(token=None)

Raise appropriate exception for HTTP errors.

Source code in src/pyetm/clients/session.py

def raise_for_status(self, token: Optional[str] = None) -> None:
    """Raise appropriate exception for HTTP errors."""
    if self.status_code == 401:
        if not token:
            raise PermissionError(
                "Authentication required: This operation requires an ETM_API_TOKEN. "
                "Set the ETM_API_TOKEN environment variable or pass a client with a token. "
                "Get your token at https://energytransitionmodel.com/api_access"
            )
        else:
            raise PermissionError(
                "Invalid ETM_API_TOKEN. Please check your token is correct and has not expired. "
                "Get a new token at https://energytransitionmodel.com/api_access"
            )

    if 400 <= self.status_code < 500:
        raise ValueError(self._format_error_message())
    if 500 <= self.status_code < 600:
        raise ConnectionError(self._format_error_message())

ETMSession

ETMSession(base_url=None, token=None, ssl_verify=None, trust_env=None, ssl_cert_path=None)

Modern async session for ETM API interactions.

Source code in src/pyetm/clients/session.py

def __init__(
    self,
    base_url: Optional[str] = None,
    token: Optional[str] = None,
    ssl_verify: Optional[bool] = None,
    trust_env: Optional[bool] = None,
    ssl_cert_path: Optional[Path] = None,
):
    self.base_url = str(base_url or get_settings().base_url).rstrip("/")
    self.token = token or get_settings().etm_api_token
    self.headers = {"Accept": "application/json"}
    if self.token:
        self.headers["Authorization"] = f"Bearer {self.token}"

    # SSL configuration
    settings = get_settings()
    self.ssl_verify = ssl_verify if ssl_verify is not None else settings.ssl_verify
    self.trust_env = trust_env if trust_env is not None else settings.trust_env
    self.ssl_cert_path = ssl_cert_path or settings.ssl_cert_path

    self._session: Optional[aiohttp.ClientSession] = None
    self._loop: Optional[asyncio.AbstractEventLoop] = None
    self._loop_thread: Optional[threading.Thread] = None
    self._loop_started = threading.Event()

    self._start_loop_thread()

Functions

request

request(method, url, **kwargs)

Make HTTP request (sync).

Source code in src/pyetm/clients/session.py

def request(self, method: str, url: str, **kwargs: Any) -> ETMResponse:
    """Make HTTP request (sync)."""
    self._ensure_session()
    if self._loop is None:
        raise RuntimeError("Event loop not initialized")
    future = asyncio.run_coroutine_threadsafe(
        self.async_request(method, url, **kwargs), self._loop
    )
    return future.result()

__getattr__

__getattr__(name)

Method generation for HTTP verbs.

Source code in src/pyetm/clients/session.py

def __getattr__(self, name: str) -> Any:
    """Method generation for HTTP verbs."""
    if name in ["get", "post", "put", "patch", "delete"]:
        return lambda url, **kwargs: self.request(name.upper(), url, **kwargs)

    if name.startswith("async_") and name[6:] in [
        "get",
        "post",
        "put",
        "patch",
        "delete",
    ]:
        method = name[6:].upper()
        return lambda url, **kwargs: self.async_request(method, url, **kwargs)

    raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")

async_request async

async_request(method, url, **kwargs)

Make async HTTP request.

Source code in src/pyetm/clients/session.py

async def async_request(self, method: str, url: str, **kwargs: Any) -> ETMResponse:
    """Make async HTTP request."""
    request_kwargs = self._build_request_kwargs(**kwargs)
    full_url = url if url.startswith("http") else f"{self.base_url}/{url.lstrip('/')}"

    if self._session is None:
        raise RuntimeError("Session not initialized")

    async with self._session.request(method, full_url, **request_kwargs) as response:
        etm_response = ETMResponse(
            status_code=response.status,
            headers=dict(response.headers),
            url=str(response.url),
        )

        content_type = response.headers.get("content-type", "").lower()

        if "application/json" in content_type:
            try:
                etm_response._json_data = await response.json()
                etm_response.text = await response.text()
            except Exception:
                etm_response.text = await response.text()
        else:
            etm_response.text = await response.text()
            etm_response._content = await response.read()

        # Check for HTTP errors after response is fully constructed
        etm_response.raise_for_status(token=self.token)

        return etm_response

close

close()

Close session and clean up resources.

Source code in src/pyetm/clients/session.py

def close(self) -> None:
    """Close session and clean up resources."""
    if self._session and self._loop:
        future = asyncio.run_coroutine_threadsafe(self._session.close(), self._loop)
        future.result()

    if self._loop:
        self._loop.call_soon_threadsafe(self._loop.stop)

Functions