biolm.core package

Subpackages

Submodules

biolm.core.asynch module

biolm.core.asynch.async_api_call_wrapper(grouped_df, slug, action, payload_maker, response_key, api_version=3, key='sequence', params=None, compress_requests=True, compress_threshold=256)

Wrap API calls to assist with sequence validation as a pre-cursor to each API call.

async biolm.core.asynch.async_api_calls(model_name, action, headers, payloads, response_key=None, api_version=3, compress_requests=True, compress_threshold=256)

Hit an arbitrary BioLM model inference API.

async biolm.core.asynch.async_main(urls, concurrency) → list

async biolm.core.asynch.async_range(count)

async biolm.core.asynch.get_all(urls: List[str], num_concurrent: int) → list

async biolm.core.asynch.get_all_biolm(url: str, ploads: List[Dict], headers: dict, num_concurrent: int, response_key: str = None, compress_requests: bool = True, compress_threshold: int = 256) → list

async biolm.core.asynch.get_one(session: ClientSession, url: str) → None

async biolm.core.asynch.get_one_biolm(session: ClientSession, url: str, pload: dict, headers: dict, response_key: str = None, compress_requests: bool = True, compress_threshold: int = 256) → None

biolm.core.auth module

biolm.core.auth.are_credentials_valid() → bool

Check if existing credentials are valid.

Returns:

True if credentials exist and are valid, False otherwise

biolm.core.auth.generate_access_token(uname, password)

Generate a TTL-expiry access and refresh token, to be used as ‘Cookie: acccess=; refresh=;” headers, or the access token only as a ‘Authorization: Bearer 1235abc’ token.

The refresh token will expire in hours or days, while the access token will have a shorter TTL, more like hours. Meaning, this method will require periodically re-logging in, due to the token expiration time. For a more permanent auth method for the API, use an API token by setting the BIOLM_TOKEN environment variable.

biolm.core.auth.get_api_token()

Get a BioLM API token to use with future API requests.

Copied from https://api.biolm.ai/#d7f87dfd-321f-45ae-99b6-eb203519ddeb.

biolm.core.auth.get_auth_status()

biolm.core.auth.get_user_auth_header()

Returns a dict with the appropriate Authorization header, either using an API token from BIOLM_TOKEN environment variable, or by reading the credentials file at ~/.biolm/credentials next.

biolm.core.auth.oauth_login(*, client_id: str | None = None, scope: str = 'read write', auth_url: str | None = None, token_url: str | None = None, redirect_uri: str | None = None) → dict

Perform OAuth login using PKCE and persist tokens to ~/.biolm/credentials.

Args:

client_id: OAuth client ID (defaults to BIOLMAI_PUBLIC_CLIENT_ID from const) scope: OAuth scope string auth_url: Authorization endpoint (defaults to OAUTH_AUTHORIZE_URL) token_url: Token endpoint (defaults to OAUTH_TOKEN_URL) redirect_uri: Redirect URI (defaults to OAUTH_REDIRECT_URI)

Returns:

Token response dict

biolm.core.auth.parse_credentials_file(file_path)

Parse credentials file, handling JSON, Python dict syntax, and mixed types.

Returns a dict with all credential fields preserved, or None if parsing fails. Uses ast.literal_eval() which is safe and only evaluates Python literals.

biolm.core.auth.refresh_access_token(refresh)

Attempt to refresh temporary user access token, by using their refresh token, which has a longer TTL.

biolm.core.auth.save_access_refresh_token(access_refresh_dict)

Save temporary access and refresh tokens to user folder for future use.

biolm.core.auth.validate_user_auth(api_token=None, access=None, refresh=None)

Validates an API token, to be used as ‘Authorization: Token 1235abc’ authentication method.

biolm.core.const module

biolm.core.const.get_base_api_url() → str

Return the model API base URL.

biolm.core.const.get_base_domain() → str

Platform site root (OAuth, auth, hosted platform APIs).

biolm.core.const.get_env_api_token() → str

Return API token from BIOLM_TOKEN, or deprecated BIOLMAI_TOKEN.

biolm.core.const.get_model_api_source() → str

Describe where the active model API URL comes from.

biolm.core.const.get_model_catalog_base() → str

Site root for model catalog/list endpoints.

When BIOLM_BASE_API_URL points at a hub or custom host, catalog listing follows that host. Platform auth still uses get_base_domain().

biolm.core.const.is_hub_mode() → bool

True when model inference targets a biolm-hub gateway (/api/v1).

biolm.core.expression_evaluator module

Template expression evaluation for protocol outputs.

This module provides safe evaluation of template expressions like ${{ field_name }} and ${{ score > 0.5 }} used in protocol output configurations.

biolm.core.expression_evaluator.evaluate_expression(expr: str, context: Dict[str, Any]) → Any

Evaluate a template expression safely.

Args:

expr: The expression to evaluate (without ${{ }} wrapper) context: Dictionary of variables available in the expression

Returns:

The evaluated result

Raises:

ValueError: If expression evaluation fails KeyError: If a required variable is missing from context

biolm.core.expression_evaluator.evaluate_template_value(value: Any, context: Dict[str, Any]) → Any

Evaluate a value that may contain a template expression.

Args:

value: Value that may be a template expression or literal context: Dictionary of variables available for evaluation

Returns:

The evaluated value (or original value if not a template expression)

biolm.core.expression_evaluator.evaluate_where_clause(expr: str, row: Dict[str, Any]) → bool

Evaluate a where clause expression row-by-row.

Args:

expr: The filter expression (may include ${{ }} wrapper) row: The row data to evaluate against

Returns:

True if row matches the filter, False otherwise

Raises:

ValueError: If expression evaluation fails

biolm.core.expression_evaluator.extract_template_expr(value: Any) → Tuple[bool, str]

Extract template expression from a value if present.

Args:

value: Value that may contain a template expression

Returns:

Tuple of (is_template_expr, expression_string) - is_template_expr: True if value is a template expression - expression_string: The inner expression (without ${{ }} wrapper)

biolm.core.http module

class biolm.core.http.AsyncRateLimiter(max_calls: int, period: float)

Bases: object

limit()

biolm.core.http.BioLMApi

alias of BlockingBioLMApi

class biolm.core.http.BioLMApiClient(model_name: str, api_key: str | None = None, base_url: str | None = None, timeout: Timeout = Timeout(connect=10.0, read=1200, write=1200, pool=1200), raise_httpx: bool = False, unwrap_single: bool = True, semaphore: int | Semaphore | None = 16, rate_limit: str | None = None, retry_error_batches: bool = True, compress_requests: bool = True, compress_threshold: int = 256, concurrent_batches: bool = True, http2: bool = True)

Bases: object

async call(func: str, items: List[dict], params: dict | None = None, raw: bool = False)

encode(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)

static extract_max_items(schema: dict) → int | None

Extracts the ‘maxItems’ value for the ‘items’ key from the schema. Returns the integer value if found, else None.

generate(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)

async lookup(query: dict | List[dict], *, raw: bool = False, output: str = 'memory', file_path: str | None = None)

predict(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True, progress_callback: Callable[[int, int], None] | None = None)

async schema(model: str, action: str) → dict | None

Fetch the JSON schema for a given model and action, with caching. Returns the schema dict if successful, else None. Uses a module-level cache keyed by (model, action) so it is safe to use from any event loop (sync wrapper, async tests, pytest-xdist workers).

Schema is always fetched from the hosted platform API without credentials. Hub gateways and custom model API hosts may not expose /schema/, and sending API tokens to the public schema endpoint can return 401.

score(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True)

search(*, items: List[dict], params: dict | None = None, stop_on_error: bool = False, output: str = 'memory', file_path: str | None = None, overwrite: bool = True)

async shutdown()

class biolm.core.http.CredentialsProvider

Bases: object

static get_auth_headers(api_key: str | None = None) → Dict[str, str]

class biolm.core.http.HttpClient(base_url: str, headers: Dict[str, str], timeout: Timeout, compress_requests: bool = True, compress_threshold: int = 256, limits: Limits | None = None, http2: bool = True)

Bases: object

async close()

Close method for backward compatibility. Shared clients are managed by _SharedClientFactory and cleaned up on process exit.

async get(endpoint: str) → Response

GET with retry logic for network errors.

async get_async_client() → AsyncClient

Get or create a shared httpx.AsyncClient instance from the factory.

async post(endpoint: str, payload: dict, extra_headers: dict | None = None) → Response

POST with optional extra_headers added just for this request.

class biolm.core.http.LookupResult(data, raw)

Bases: tuple

data

Alias for field number 0

raw

Alias for field number 1

biolm.core.http.custom_httpx_encode_json(json: Any) → Tuple[Dict[str, str], ByteStream]

biolm.core.http.debug(msg)

biolm.core.http.parse_rate_limit(rate: str)

biolm.core.http.type_check(param_types: Dict[str, Any])

biolm.core.paths module

Canonical user config paths under ~/.biolm with ~/.biolmai fallbacks.

biolm.core.paths.ensure_user_config_dir() → Path

Create and return the canonical user config directory.

biolm.core.paths.legacy_user_config_dir() → Path

Return the deprecated user config directory (~/.biolmai).

biolm.core.paths.resolve_user_file(relative_path: str | Path, *, warn_legacy: bool = True) → str

Like resolve_user_path() but returns a string path.

biolm.core.paths.resolve_user_path(relative_path: str | Path, *, warn_legacy: bool = True) → Path

Resolve a path under the user config dir with legacy fallback.

If the canonical path exists, it is returned. Otherwise, if the legacy path exists, it is returned after emitting a deprecation warning. If neither exists, the canonical path is returned (for new writes).

biolm.core.paths.user_config_dir() → Path

Return the canonical user config directory (~/.biolm).

biolm.core.paths.warn_deprecated_path(legacy_path: str | Path, canonical_path: str | Path) → None

Emit a DeprecationWarning once per legacy path.

biolm.core.payloads module

biolm.core.payloads.INST_DAT_TXT(batch, include_batch_size=False)

biolm.core.payloads.PARAMS_ITEMS(batch, key='sequence', params=None, include_batch_size=False)

biolm.core.payloads.predict_resp_many_in_one_to_many_singles(resp_json, status_code, batch_id, local_err, batch_size, response_key='results')

biolm.core.seqflow_auth module

MLflow RequestHeaderProvider that uses biolm credentials for authentication.

This provider reads OAuth tokens from ~/.biolm/credentials and adds them as Bearer tokens to MLflow requests.

class biolm.core.seqflow_auth.BiolmRequestHeaderProvider

Bases: BiolmaiRequestHeaderProvider

Canonical MLflow header provider name after biolm rename.

class biolm.core.seqflow_auth.BiolmaiRequestHeaderProvider

Bases: object

MLflow RequestHeaderProvider that uses biolm credentials (deprecated class name).

Reads OAuth tokens from ~/.biolm/credentials (JSON format: {“access”: “…”, “refresh”: “…”}) and adds Authorization header with Bearer token to all MLflow requests.

in_context() → bool

Return True if provider should be used.

Checks if: 1. biolm package is installed 2. Credentials file exists 3. Credentials file contains access token

request_headers() → dict

Return headers to add to MLflow requests.

Returns Authorization header with Bearer token from biolm credentials.

biolm.core.utils module

Utility functions for BioLM SDK.

biolm.core.utils.batch_iterable(iterable, batch_size)

biolm.core.utils.is_list_of_lists(items, check_n=10)

biolm.core.utils.prepare_items_for_api(items: Any | List[Any], type: str | None = None, check_n: int = 10) → Tuple[List[dict] | List[List[dict]] | Any, bool]

Normalize items for API calls. Supports list, tuple, single value, and iterables (e.g. generators). Returns (data, is_lol) where is_lol is True for list-of-lists; data is iterable of dicts or list of lists.

biolm.core.validate module

class biolm.core.validate.AAExtended

Bases: object

class biolm.core.validate.AAExtendedPlusExtra(extra: List[str])

Bases: object

class biolm.core.validate.AAUnambiguous

Bases: object

class biolm.core.validate.AAUnambiguousEmpty

Bases: object

class biolm.core.validate.AAUnambiguousPlusExtra(extra: List[str])

Bases: object

class biolm.core.validate.DNAUnambiguous

Bases: object

class biolm.core.validate.PDB

Bases: object

class biolm.core.validate.SingleOccurrenceOf(single_token: str)

Bases: object

class biolm.core.validate.SingleOrMoreOccurrencesOf(token: str)

Bases: object

biolm.core.validate.aa_extended_validator(text: str) → str

biolm.core.validate.aa_unambiguous_validator(text: str) → str

biolm.core.validate.dna_unambiguous_validator(text: str) → str

biolm.core.validate.empty_or_aa_unambiguous_validator(text: str) → str

biolm.core.validate.empty_or_dna_unambiguous_validator(text: str) → str

biolm.core.validate.pdb_validator(text: str) → str

Module contents

Core package for BioLM SDK.