How to use oaa-runner utility modules

oaa-runner ships two utility modules used in custom fetch modules: params_or_env for resolving configuration values, and BaseURLSession for making HTTP requests against a fixed base URL. Both are in oaa.modules.

params_or_env

from oaa.modules.params_or_env import params_or_env

Resolves a configuration value from up to four sources, in priority order:

  1. Environment variable (highest priority)

  2. connection.params[key]

  3. source.params[key]

  4. default argument (lowest priority, None if omitted)

class MyAPI:
    def __init__(self, connection, source):
        self.api_key = params_or_env(connection, source, 'MY_API_KEY')
        self.base_url = params_or_env(connection, source, 'base_url')
        self.timeout = params_or_env(connection, source, 'timeout', default=30)

Why environment variables have highest priority: This lets operators override config values at runtime without editing config.yaml. A key in connection.params sets the default; an env var with the same name overrides it.

Typical usage in tutorials: Credential keys (BW_APP_CLIENT_ID) are read from env vars. Infrastructure keys (base_url, dataset_id) are read from connection.params. Both use the same params_or_env call — the lookup order handles the routing.

BaseURLSession

from oaa.modules.base_url_session import BaseURLSession

A requests.Session subclass that prepends a base URL to every request. Eliminates repeated string construction and ensures consistent URL joining.

session = BaseURLSession(base_url='https://api.example.com/v1')
session.headers.update({'Authorization': 'Bearer token'})

response = session.get('/users')       # GET https://api.example.com/v1/users
response = session.get('/groups/42')   # GET https://api.example.com/v1/groups/42

Leading slashes on relative paths are stripped before joining — both /users and users produce the same URL. The base URL always ends with / (added automatically if missing).

Typical pattern in custom modules:

@property
def session(self):
    if not self._session:
        self._session = BaseURLSession(base_url=f"{self.base_url}/{self.dataset_id}")
    self._session.headers.update({'Authorization': f'Bearer {self._access_token}'})
    return self._session

See also