Connections & OAuth¶

A connection holds the credentials a source authenticates with; the OAuth variants renew them through a provider.

connection ¶

connection(target: Any = None, /, **settings: Any) -> Any

Create a Connection subclass from a decorated class.

Can be used bare or with arguments:

@connection
class MyConnection:
    host: str = "localhost"
    port: int = 5432
    username: str
    password: str

@connection(key="custom", name="Custom Connection")
class OtherConnection:
    url: str

Class-level traits, identity (key, name, icon, tags) and behaviour (oauth), belong in the decorator; the class body declares fields:

@connection(
    name="Amazon Ads",
    oauth=OAuthConfig("amazon", scope="advertising::campaign_management"),
)
class AmazonAdsConnection(OAuthConnection):
    location: str = SelectField(...)

The decorated class's annotations and attributes become the Connection subclass body. Since Connection extends BaseSettings, fields can still be loaded from environment variables.

relations maps a relation name to a Relation, a component class, or a list of component classes narrowing the relation the decorated class declares under that name. Every other keyword is one of the public ClassVars and field defaults the decorated class declares, or Connection itself for a plain class (key, name, icon, tags, auto_renew, and oauth on an OAuthConnection). An unknown name is a TypeError at decoration, naming the subclass that declares it when one does. Building the class fails with a TypeError if oauth.fields maps token response keys to model fields the class does not declare.

Connection ¶

Connection(**data: Any)

Bases: interloper.resource.base.Resource, interloper.operation.base.Operation

A resource for database/service connection credentials.

Like every Resource, connection values can be loaded from environment variables, .env files, or passed directly:

class MyConnection(Connection):
    host: str = "localhost"
    port: int = 5432
    username: str
    password: str

# Loads USERNAME, PASSWORD from environment if not passed explicitly
connection = MyConnection()

Connections that support OAuth subclass OAuthConnection (custom token shape) or RefreshTokenOAuthConnection (standard refresh-token trio); the oauth config lives there, not on this base.

billable ¶

billable = False

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

capture_traceback ¶

capture_traceback = False

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

enabled ¶

enabled = <interloper.component.inherited.InheritedSetting object at 0x7febf2273350>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

internal_fields ¶

internal_fields = frozenset({'enabled'})

frozenset() -> empty frozenset object frozenset(iterable) -> frozenset object

Build an immutable unordered collection of unique elements.

key ¶

key = 'connection'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

kind ¶

kind = 'connection'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'extra': 'forbid', 'arbitrary_types_allowed': True, 'validate_default': True, 'case_sensitive': False, 'env_prefix': '', 'env_prefix_target': 'variable', 'nested_model_default_partial_update': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_none_str': None, 'env_parse_enums': None, 'cli_prog_name': None, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_hide_none_type': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_use_class_docs_for_groups': False, 'cli_show_env_vars': False, 'cli_exit_on_error': True, 'cli_prefix': '', 'cli_flag_prefix_char': '-', 'cli_implicit_flags': False, 'cli_ignore_unknown_args': False, 'cli_kebab_case': False, 'cli_shortcuts': None, 'json_file': None, 'json_file_encoding': None, 'yaml_file': None, 'yaml_file_encoding': None, 'yaml_config_section': None, 'toml_file': None, 'secrets_dir': None, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'enable_decoding': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>, <class 'interloper.component.relation.Relation'>)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

relations ¶

relations = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

renewal_interval ¶

renewal_interval = datetime.timedelta(days=1)

Difference between two datetime values.

timedelta(days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0, hours=0, weeks=0)

All arguments are optional and default to 0. Arguments may be integers or floats, and may be positive or negative.

retry ¶

retry = <interloper.component.inherited.InheritedSetting object at 0x7febf2270920>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

tags ¶

tags = []

Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

state_model ¶

Bases: pydantic.main.BaseModel

Machine-owned connection state (see Component.state_model).

Written by the renewal pipeline: the renewal controller stamps a provisional next_renewal_at when it enqueues a renewal run, and the run overwrites it with the real next due time on completion. Timestamps are canonical timezone-aware ISO-8601 strings, compared lexicographically in SQL like JobState's. last_run_at and last_run_status are stamped by run completion on every run's component — plumbing, hidden from state columns.

model_config ¶

model_config = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

check ¶

check() -> bool

Verify this connection with a lightweight authenticated call.

Override in a subclass (sync or async) to make the cheapest call that proves the credentials work — often a one-line delegation to an existing @fetch_field_provider method. Like fetch providers, it may run inside the API process, so it must use lightweight HTTP (httpx2), never a heavy provider SDK.

Returns True when the connection works, False when it provably doesn't. Exceptions are also failures: httpx2 errors are categorised by the caller (401/403 → bad credentials, timeouts → network); raise ConnectionCheckError to surface a curated message instead.

checkable ¶

checkable() -> bool

Whether this connection class implements check.

Returns:

Type	Description
`bool`	True when the class overrides the base hook.

definition ¶

definition() -> ResourceDefinition

Advertise check / renew support so UIs can offer them.

Returns:

Type	Description
`ResourceDefinition`	The resource definition with `checkable` and `renewable` set.
`ResourceDefinition`	`auto_renew` and `retry` are dropped from the config schema
`ResourceDefinition`	when the class has nothing to renew: both govern the renewal and
`ResourceDefinition`	would be inert.

execute ¶

execute(context: OperationContext) -> OperationResult

Renew this connection's credentials: the connection's operation.

The template over renew: caps the exchange at _RENEWAL_TIMEOUT, returns any rotated credential fields as config effects, and stamps the next due time — half the reported validity when the provider gives one (floored at 15 minutes), the class's renewal_interval otherwise.

Parameters:

Name	Type	Description	Default
`context`	`OperationContext`	The facts this execution is scoped to, unused — a renewal needs nothing beyond the connection itself.	required

Returns:

Type	Description
`OperationResult`	The renewal's effects.

failure ¶

failure(error: Exception) -> OperationResult

Describe a failed renewal: a curated message plus a retry slot.

The message comes from renewal_failure_message, so raw provider errors (which embed credentials in URLs) never reach the connection's state or the run's failure event; the retry slot makes the connection due again without waiting a full interval.

Parameters:

Name	Type	Description	Default
`error`	`Exception`	The exception `execute` raised.	required

Returns:

Type	Description
`OperationResult`	A failed result stamping the curated error and the retry slot.

model_post_init ¶

model_post_init(context: Any) -> None

Default id to a generated UUID if not provided.

Parameters:

Name	Type	Description	Default
`context`	`Any`	The pydantic validation context, unused here.	required

renew ¶

renew() -> Renewal

Renew this connection's credentials before they age out.

Override in a subclass (sync or async) to exchange the stored credential for a fresh one — same execution contract as check: lightweight HTTP (httpx2), never a heavy provider SDK. OAuth connections need no override: OAuthConnection derives the whole flow from the provider and the oauth.fields mapping.

The renewal pipeline persists the returned Renewal.fields into the stored config and schedules the next renewal; a raised exception marks the renewal failed (surfaced on the connection's state), typically meaning the credential is dead and a human must re-consent.

renewable ¶

renewable() -> bool

Whether this connection class implements renew.

Returns:

Type	Description
`bool`	True when the class overrides the base hook.

renewal_failure_message ¶

renewal_failure_message(error: Exception) -> str

Describe a failed renewal in terms that are safe to persist.

Provider token exchanges carry credentials in URLs and bodies, and httpx2 error strings embed the request URL — so raw messages must never reach the connection's state or the run's error event. HTTP-layer failures collapse to their category; anything else formats through format_exception (which already strips pydantic input values).

Parameters:

Name	Type	Description	Default
`error`	`Exception`	The exception `renew` raised.	required

Returns:

Type	Description
`str`	A short, curated message.

ConnectionState ¶

Bases: pydantic.main.BaseModel

Machine-owned connection state (see Component.state_model).

Written by the renewal pipeline: the renewal controller stamps a provisional next_renewal_at when it enqueues a renewal run, and the run overwrites it with the real next due time on completion. Timestamps are canonical timezone-aware ISO-8601 strings, compared lexicographically in SQL like JobState's. last_run_at and last_run_status are stamped by run completion on every run's component — plumbing, hidden from state columns.

model_config ¶

model_config = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

OAuthConnection ¶

OAuthConnection(**data: Any)

Bases: interloper.connection.base.Connection

OAuth connection base — drives the "Sign in with X" form from oauth.

Carries the oauth config but no credential fields. Subclass this directly when the connection's credential fields are named differently from the standard trio, declaring its own fields and a matching OAuthConfig.fields mapping:

@connection(oauth=OAuthConfig("tiktok", fields={"refresh_token": "access_token"}))
class TiktokAdsConnection(OAuthConnection):
    access_token: str = SecretField()

OAuthConfig.fields maps the OAuth roles (client_id / client_secret / refresh_token) to the connection's field names, and drives the whole form: those fields are hidden in sign-in mode, the client_id / client_secret fields are resolved from env, and the refresh_token field receives the token from the sign-in response.

Most connections use the standard refresh-token flow and should subclass RefreshTokenOAuthConnection instead, which declares the trio.

enabled ¶

enabled = <interloper.component.inherited.InheritedSetting object at 0x7febf2279190>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

key ¶

key = 'o_auth_connection'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'extra': 'forbid', 'arbitrary_types_allowed': True, 'validate_default': True, 'case_sensitive': False, 'env_prefix': '', 'env_prefix_target': 'variable', 'nested_model_default_partial_update': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_none_str': None, 'env_parse_enums': None, 'cli_prog_name': None, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_hide_none_type': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_use_class_docs_for_groups': False, 'cli_show_env_vars': False, 'cli_exit_on_error': True, 'cli_prefix': '', 'cli_flag_prefix_char': '-', 'cli_implicit_flags': False, 'cli_ignore_unknown_args': False, 'cli_kebab_case': False, 'cli_shortcuts': None, 'json_file': None, 'json_file_encoding': None, 'yaml_file': None, 'yaml_file_encoding': None, 'yaml_config_section': None, 'toml_file': None, 'secrets_dir': None, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'enable_decoding': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>, <class 'interloper.component.relation.Relation'>)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

relations ¶

relations = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

retry ¶

retry = <interloper.component.inherited.InheritedSetting object at 0x7febf2273f20>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

definition ¶

definition() -> ResourceDefinition

Inject the x-oauth extension so the form can render the button.

The extension carries the provider metadata and the fields mapping; the form uses fields to hide the credential fields in sign-in mode and to place the token returned by the flow.

Returns:

Type	Description
`ResourceDefinition`	The resource definition, enriched when `oauth` is configured.

env_credential ¶

env_credential(suffix: str) -> str | None

The in-house OAuth credential for suffix (e.g. "CLIENT_ID").

Read from INTERLOPER_<PROVIDER>_<SUFFIX> — the same vars the token-exchange endpoint uses. A mode="before" validator falls back to this for its declared credential fields, so a required field can be satisfied by the in-house app. Returns None when unset, so a field left unfilled stays None and fails the required check rather than passing as an empty string.

Parameters:

Name	Type	Description	Default
`suffix`	`str`	Credential field suffix — `CLIENT_ID`, `CLIENT_SECRET` or `REDIRECT_URI`; case-insensitive.	required

Returns:

Type	Description
`str \| None`	The env credential value, or `None`.

model_post_init ¶

model_post_init(context: Any) -> None

Default id to a generated UUID if not provided.

Parameters:

Name	Type	Description	Default
`context`	`Any`	The pydantic validation context, unused here.	required

renew ¶

renew() -> Renewal

Renew the stored credential through the provider's refresh flow.

One implementation serves every OAuth connection, driven entirely by what is already declared: the registered provider builds the grant and parses its response (dialect included — Facebook's fb_exchange_token, Microsoft's scoped refresh), and oauth.fields names which connection fields play the grant's credential roles, the rotated credential landing back in the field playing the refresh-token role. Renewing keeps inactivity-expiring credentials alive and surfaces a dead one as a failure instead of a later run error.

Returns:

Type	Description
`Renewal`	The rotated credential field when the provider issued a new one,
`Renewal`	and the credential's validity when reported.

Raises:

Type	Description
`NotImplementedError`	When the connection is not renewable (see `renewable`).

renewable ¶

renewable() -> bool

Whether this connection can renew — derived, never hand-written.

A bespoke renew override always counts. Otherwise renewability follows from the declarations alone: the registered provider must have a refresh flow (supports_refresh) and oauth.fields must name a connection field for each of the grant's credential roles. A qualifying connection renews with zero renewal code, and one whose provider has no refresh flow (TikTok: tokens do not expire) is excluded automatically.

Returns:

Type	Description
`bool`	True when renewal is overridden or fully derivable.

resolve_field ¶

resolve_field(
    data: dict[str, Any], field: str, value: str | None
) -> None

Fill data[field] from value when the field is blank and value is set.

A mode="before" resolve_credentials validator uses this per field. Leaves the field absent when there is no value, so a required field fails with "Field required" rather than being satisfied by an empty value.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The raw pre-validation input, mutated in place.	required
`field`	`str`	Name of the model field to fill.	required
`value`	`str \| None`	The value to place, or `None` when nothing was resolved.	required

RefreshTokenOAuthConnection ¶

RefreshTokenOAuthConnection(**data: Any)

Bases: interloper.connection.base.OAuthConnection

A connection using the standard OAuth2 refresh-token flow.

Declares the credential trio — client_id / client_secret (in-house OAuth credentials resolved from env) and refresh_token (filled on sign-in) — which the default OAuthConfig.fields mapping targets, so the standard connections only add their own fields:

@connection(oauth=OAuthConfig("linkedin", scope="r_ads"))
class LinkedinAdsConnection(RefreshTokenOAuthConnection):
    account_id: str

Connections whose credential fields are named differently subclass OAuthConnection directly, declare their own fields, and inject them from env with a mode="before" validator calling env_credential.

All three fields are required. client_id / client_secret may be supplied by the in-house per-provider credentials (INTERLOPER_<PROVIDER>_CLIENT_ID / INTERLOPER_<PROVIDER>_CLIENT_SECRET — the same vars the token-exchange endpoint reads), injected before validation so the sign-in flow can omit them and the in-house secret is never sent to the browser or stored per connection. An explicit value overrides the in-house app; when neither the caller nor env supplies one, the required check fails.

enabled ¶

enabled = <interloper.component.inherited.InheritedSetting object at 0x7febf21e5a90>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

key ¶

key = 'refresh_token_o_auth_connection'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

model_config ¶

model_config = {'extra': 'forbid', 'arbitrary_types_allowed': True, 'validate_default': True, 'case_sensitive': False, 'env_prefix': '', 'env_prefix_target': 'variable', 'nested_model_default_partial_update': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_none_str': None, 'env_parse_enums': None, 'cli_prog_name': None, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_hide_none_type': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_use_class_docs_for_groups': False, 'cli_show_env_vars': False, 'cli_exit_on_error': True, 'cli_prefix': '', 'cli_flag_prefix_char': '-', 'cli_implicit_flags': False, 'cli_ignore_unknown_args': False, 'cli_kebab_case': False, 'cli_shortcuts': None, 'json_file': None, 'json_file_encoding': None, 'yaml_file': None, 'yaml_file_encoding': None, 'yaml_config_section': None, 'toml_file': None, 'secrets_dir': None, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'enable_decoding': True, 'ignored_types': (<class 'interloper.serializable.base.IgnoredDescriptor'>, <class 'interloper.component.relation.Relation'>)}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

relations ¶

relations = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

retry ¶

retry = <interloper.component.inherited.InheritedSetting object at 0x7febf21e5a30>

The read-through descriptor installed over one inherited field.

A data descriptor, so it wins over the instance __dict__ on reads while Pydantic keeps writing there: the stored value is the override, and the read is the effective value.

model_post_init ¶

model_post_init(context: Any) -> None

Default id to a generated UUID if not provided.

Parameters:

Name	Type	Description	Default
`context`	`Any`	The pydantic validation context, unused here.	required

resolve_credentials ¶

resolve_credentials(data: Any) -> Any

Inject blank client_id / client_secret from the in-house env creds.

Runs before validation so the in-house app can satisfy these required fields when the caller omits them; an explicit value is left untouched.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The raw pre-validation input; only a `dict` is augmented, anything else passes through untouched.	required

Returns:

Type	Description
`Any`	The (possibly augmented) input data.

Renewal ¶

Bases: pydantic.main.BaseModel

The outcome of a successful Connection.renew.

fields maps connection field names to their renewed values — the platform persists them into the stored (encrypted) config, so a rotated or re-issued credential replaces the aging one. Empty means the renewal succeeded but nothing needs persisting (a pure keep-alive).

expires_in is the renewed credential's validity in seconds when the provider reports one; the platform schedules the next renewal from it (with a safety margin), falling back to the class's renewal_interval when absent.

model_config ¶

model_config = {}

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

OAuthProvider ¶

OAuthProvider(
    key: str,
    auth_url: str,
    token_url: str,
    label: str = "",
    icon: str = "",
    token_encoding: Literal["json", "form"] = "json",
)

Identity and token-flow dialect of an OAuth2 provider.

The base class speaks plain RFC 6749: the authorization-code exchange and the refresh-token grant are POSTs to token_url with the standard parameters. A provider whose dialect deviates — a different method, parameter names, an extra parameter, another grant entirely, a response envelope — overrides the request builders and response parsers on a subclass; the provider is the single owner of its dialect, and the sign-in exchange and connection renewal consume it identically.

token_encoding is the one wire knob a plain instance may set, and it only parameterizes the default builders' body encoding. Any other deviation is a method override, never a new field. supports_refresh is a class trait of the dialect: a provider with no credential-refresh flow at all (TikTok: tokens do not expire) sets it False on its subclass, and connections on it derive as non-renewable.

Parameters:

Name	Type	Description	Default
`key`	`str`	Provider key (e.g. `"amazon"`) — the registry key.	required
`auth_url`	`str`	Authorization endpoint the user is sent to.	required
`token_url`	`str`	Token endpoint the token requests are sent to.	required
`label`	`str`	Display label (defaults to the titlecased key).	`''`
`icon`	`str`	Icon identifier (e.g. `"logos:facebook"`).	`''`
`token_encoding`	`Literal['json', 'form']`	Body encoding of the default token requests.	`'json'`

icon ¶

icon = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

label ¶

label = ''

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

supports_refresh ¶

supports_refresh = True

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

token_encoding ¶

token_encoding = 'json'

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

authorization_code_request ¶

authorization_code_request(
    *,
    code: str,
    redirect_uri: str,
    client_id: str,
    client_secret: str,
) -> Request

Build the authorization-code grant's token request (RFC 6749 §4.1.3).

Parameters:

Name	Type	Description	Default
`code`	`str`	The authorization code from the provider's consent screen.	required
`redirect_uri`	`str`	The redirect URI the code was issued against.	required
`client_id`	`str`	The OAuth app's client id.	required
`client_secret`	`str`	The OAuth app's client secret.	required

Returns:

Type	Description
`Request`	The request to send; the caller owns the client and error handling.

parse_authorization_code_response ¶

parse_authorization_code_response(
    payload: dict[str, Any],
) -> dict[str, Any]

Read the authorization-code grant's response into the RFC token shape.

The default relays the body as is: RFC 6749 §5.1 already puts access_token / refresh_token at the top level. A provider that wraps them in an envelope, or reports rejections inside a 200 body, overrides this to unwrap and to raise.

Parameters:

Name	Type	Description	Default
`payload`	`dict[str, Any]`	The grant's JSON response body.	required

Returns:

Type	Description
`dict[str, Any]`	The token response with the credentials at the top level.

parse_refresh_token_response ¶

parse_refresh_token_response(
    payload: dict[str, Any],
) -> RefreshTokenResponse

Read the refresh grant's response at the credential-role level.

The default reads the RFC shape: refresh_token is the (possibly rotated) credential when present, and refresh_token_expires_in its validity — expires_in describes the access token, which interloper does not store.

Parameters:

Name	Type	Description	Default
`payload`	`dict[str, Any]`	The grant's JSON response body.	required

Returns:

Type	Description
`RefreshTokenResponse`	The issued refresh credential and its validity, when reported.

refresh_token_request ¶

refresh_token_request(
    *,
    client_id: str,
    client_secret: str,
    refresh_token: str,
    scope: str | None = None,
) -> Request

Build the refresh-token grant request (RFC 6749 §6).

The default grant omits scope even when given: it is optional per the RFC and some providers reject parameters they do not document. A provider that requires it overrides (Microsoft: AADSTS90023).

Parameters:

Name	Type	Description	Default
`client_id`	`str`	The OAuth app's client id.	required
`client_secret`	`str`	The OAuth app's client secret.	required
`refresh_token`	`str`	The credential to exchange for a fresh one.	required
`scope`	`str \| None`	The connection's declared scope; ignored by the default grant.	`None`

Returns:

Type	Description
`Request`	The request to send; the caller owns the client and error handling.

OAuthConfig ¶

OAuthConfig(
    provider: str,
    *,
    scope: str = "",
    fields: dict[str, str] | None = None,
    auth_url: str = "",
    label: str = "",
    icon: str = "",
)

Class-level OAuth configuration for connection resources.

Declares that a connection supports OAuth sign-in: which provider to use and what scope to request. Provider identity (auth_url, label, icon) is resolved from the provider registry; pass them explicitly only to override the registry or to use an unregistered provider.

When present on a connection class (via the @connection(oauth=...) decorator kwarg or an oauth ClassVar), definition() injects an x-oauth extension into the root of the generated JSON Schema so the frontend can render a "Sign in with X" button.

fields maps the canonical OAuth roles — client_id / client_secret / refresh_token — to the connection's actual field names. It drives the whole form: the mapped fields are hidden in sign-in mode, the client_id / client_secret fields are resolved from INTERLOPER_<PROVIDER>_CLIENT_ID / INTERLOPER_<PROVIDER>_CLIENT_SECRET env, and the refresh_token field receives the token from the sign-in response. It defaults to the identity trio, so a standard connection declares nothing; a connection with differently named fields (or only some roles) passes its own mapping.

Parameters:

Name	Type	Description	Default
`provider`	`str`	OAuth provider key (e.g. `"amazon"`, `"facebook"`).	required
`scope`	`str`	OAuth scope string to request.	`''`
`fields`	`dict[str, str] \| None`	`{role: model_field_name}` mapping (roles: `client_id` / `client_secret` / `refresh_token`). Defaults to the identity trio.	`None`
`auth_url`	`str`	Authorization endpoint override. Required when the provider is not in the registry.	`''`
`label`	`str`	Display label override.	`''`
`icon`	`str`	Icon identifier override.	`''`

Raises:

Type	Description
`ConfigError`	If the provider is unregistered and no `auth_url` is given.

Example:

@connection(
    name="Facebook Ads",
    oauth=OAuthConfig(
        "facebook",
        scope="ads_read",
        fields={"client_id": "app_id", "client_secret": "app_secret", "refresh_token": "access_token"},
    ),
)
class FacebookAdsConnection(OAuthConnection):
    ...

Initialise an OAuthConfig, resolving provider metadata from the registry.

Parameters:

Name	Type	Description	Default
`provider`	`str`	OAuth provider key (e.g. `"amazon"`).	required
`scope`	`str`	OAuth scope string.	`''`
`fields`	`dict[str, str] \| None`	Role → field-name mapping (see class docstring); defaults to the identity trio.	`None`
`auth_url`	`str`	Authorization endpoint override (required for unregistered providers).	`''`
`label`	`str`	Display label override.	`''`
`icon`	`str`	Icon identifier override.	`''`

Raises:

Type	Description
`ConfigError`	If the provider is unregistered and no `auth_url` is given.

to_schema_ext ¶

to_schema_ext() -> dict[str, Any]

Serialize to a JSON Schema x-oauth extension dict.

Returns:

Type	Description
`dict[str, Any]`	Dict with provider, auth_url, scope, label, icon, and fields.

RefreshTokenResponse ¶

RefreshTokenResponse(
    refresh_token: str | None = None,
    expires_in: int | None = None,
)

Role-level outcome of a refresh-token grant.

Parameters:

Name	Type	Description	Default
`refresh_token`	`str \| None`	The refresh credential the provider issued, or `None` when the response carried none (no rotation).	`None`
`expires_in`	`int \| None`	Validity of the refresh credential in seconds, when the provider reports it.	`None`