Submodules

airbyte_cdk.sources.declarative.interpolation.interpolated_boolean module

class airbyte_cdk.sources.declarative.interpolation.interpolated_boolean.InterpolatedBoolean(condition: str, parameters: dataclasses.InitVar[Mapping[str, Any]])

Bases: object

condition: str

eval(config: Mapping[str, Any], **additional_parameters)

Interpolates the predicate condition string using the config and other optional arguments passed as parameter.

Parameters

• config – The user-provided configuration as specified by the source’s spec

• additional_parameters – Optional parameters used for interpolation

Returns

The interpolated string

parameters: dataclasses.InitVar[Mapping[str, Any]]

airbyte_cdk.sources.declarative.interpolation.interpolated_mapping module

class airbyte_cdk.sources.declarative.interpolation.interpolated_mapping.InterpolatedMapping(mapping: Mapping[str, str], parameters: dataclasses.InitVar[Mapping[str, Any]])

Bases: object

Wrapper around a Mapping[str, str] where both the keys and values are to be interpolated.

mapping

to be evaluated

Type

Mapping[str, str]

eval(config: Mapping[str, Any], **additional_parameters)

Wrapper around a Mapping[str, str] that allows for both keys and values to be interpolated.

Parameters

• config – The user-provided configuration as specified by the source’s spec

• additional_parameters – Optional parameters used for interpolation

Returns

The interpolated string

mapping: Mapping[str, str]

parameters: dataclasses.InitVar[Mapping[str, Any]]

airbyte_cdk.sources.declarative.interpolation.interpolated_string module

class airbyte_cdk.sources.declarative.interpolation.interpolated_string.InterpolatedString(string: str, parameters: dataclasses.InitVar[Mapping[str, Any]], default: Optional[str] = None)

Bases: object

Wrapper around a raw string to be interpolated with the Jinja2 templating engine

string

The string to evalute

Type

str

default

The default value to return if the evaluation returns an empty string

Type

Optional[str]

parameters

Additional runtime parameters to be used for string interpolation

Type

Mapping[str, Any]

classmethod create(string_or_interpolated: Union[airbyte_cdk.sources.declarative.interpolation.interpolated_string.InterpolatedString, str], *, parameters: Mapping[str, Any])

Helper function to obtain an InterpolatedString from either a raw string or an InterpolatedString.

Parameters

• string_or_interpolated – either a raw string or an InterpolatedString.

• parameters – parameters propagated from parent component

Returns

InterpolatedString representing the input string.

default: Optional[str] = None

eval(config: Mapping[str, Any], **kwargs)

Interpolates the input string using the config and other optional arguments passed as parameter.

Parameters

• config – The user-provided configuration as specified by the source’s spec

• kwargs – Optional parameters used for interpolation

Returns

The interpolated string

parameters: dataclasses.InitVar[Mapping[str, Any]]

string: str

airbyte_cdk.sources.declarative.interpolation.interpolation module

class airbyte_cdk.sources.declarative.interpolation.interpolation.Interpolation

Bases: abc.ABC

Strategy for evaluating the interpolated value of a string at runtime using Jinja.

abstract eval(input_str: str, config: Mapping[str, Any], default: Optional[str] = None, **additional_options)

Interpolates the input string using the config, and additional options passed as parameter.

Parameters

• input_str – The string to interpolate

• config – The user-provided configuration as specified by the source’s spec

• default – Default value to return if the evaluation returns an empty string

• additional_options – Optional parameters used for interpolation

Returns

The interpolated string

airbyte_cdk.sources.declarative.interpolation.jinja module

class airbyte_cdk.sources.declarative.interpolation.jinja.JinjaInterpolation

Bases: airbyte_cdk.sources.declarative.interpolation.interpolation.Interpolation

Interpolation strategy using the Jinja2 template engine.

If the input string is a raw string, the interpolated string will be the same. eval(“hello world”) -> “hello world”

The engine will evaluate the content passed within {{}}, interpolating the keys from the config and context-specific arguments. eval(“hello {{ name }}”, name=”airbyte”) -> “hello airbyte”) eval(“hello {{ config.name }}”, config={“name”: “airbyte”}) -> “hello airbyte”)

In additional to passing additional values through the kwargs argument, macros can be called from within the string interpolation. For example, “{{ max(2, 3) }}” will return 3

Additional information on jinja templating can be found at https://jinja.palletsprojects.com/en/3.1.x/templates/#

ALIASES = {'stream_interval': 'stream_slice', 'stream_partition': 'stream_slice'}

RESTRICTED_BUILTIN_FUNCTIONS = ['range']

RESTRICTED_EXTENSIONS = ['jinja2.ext.loopcontrols']

eval(input_str: str, config: Mapping[str, Any], default: Optional[str] = None, valid_types: Optional[Tuple[Type[Any]]] = None, **additional_parameters)

Interpolates the input string using the config, and additional options passed as parameter.

Parameters

• input_str – The string to interpolate

• config – The user-provided configuration as specified by the source’s spec

• default – Default value to return if the evaluation returns an empty string

• additional_options – Optional parameters used for interpolation

Returns

The interpolated string

airbyte_cdk.sources.declarative.interpolation.macros module

airbyte_cdk.sources.declarative.interpolation.macros.day_delta(num_days: int, format: str = '%Y-%m-%dT%H:%M:%S.%f%z') → str

Returns datetime of now() + num_days

Usage: “{{ day_delta(25) }}”

Parameters

num_days – number of days to add to current date time

Returns

datetime formatted as RFC3339

airbyte_cdk.sources.declarative.interpolation.macros.duration(datestring: str) → datetime.timedelta

Converts ISO8601 duration to datetime.timedelta

Usage: “{{ now_utc() - duration(‘P1D’) }}”

airbyte_cdk.sources.declarative.interpolation.macros.format_datetime(dt: Union[str, datetime.datetime], format: str) → str

Converts datetime to another format

Usage: “{{ format_datetime(config.start_date, ‘%Y-%m-%d’) }}”

airbyte_cdk.sources.declarative.interpolation.macros.max(*args)

Returns biggest object of an iterable, or two or more arguments.

max(iterable, *[, default=obj, key=func]) -> value max(arg1, arg2, *args, *[, key=func]) -> value

Usage: `”{{ max(2,3) }}”

With a single iterable argument, return its biggest item. The default keyword-only argument specifies an object to return if the provided iterable is empty. With two or more arguments, return the largest argument. :param args: args to compare :return: largest argument

airbyte_cdk.sources.declarative.interpolation.macros.now_utc()

Current local date and time in UTC timezone

Usage: “{{ now_utc() }}”

airbyte_cdk.sources.declarative.interpolation.macros.timestamp(dt: Union[numbers.Number, str])

Converts a number or a string to a timestamp

If dt is a number, then convert to an int If dt is a string, then parse it using dateutil.parser

Usage: `”{{ timestamp(1658505815.223235) }}”

Parameters

dt – datetime to convert to timestamp

Returns

unix timestamp

airbyte_cdk.sources.declarative.interpolation.macros.today_utc()

Current date in UTC timezone

Usage: “{{ today_utc() }}”

Module contents

class airbyte_cdk.sources.declarative.interpolation.InterpolatedBoolean(condition: str, parameters: dataclasses.InitVar[Mapping[str, Any]])

Bases: object

condition: str

eval(config: Mapping[str, Any], **additional_parameters)

Interpolates the predicate condition string using the config and other optional arguments passed as parameter.

Parameters

• config – The user-provided configuration as specified by the source’s spec

• additional_parameters – Optional parameters used for interpolation

Returns

The interpolated string

parameters: dataclasses.InitVar[Mapping[str, Any]]

class airbyte_cdk.sources.declarative.interpolation.InterpolatedMapping(mapping: Mapping[str, str], parameters: dataclasses.InitVar[Mapping[str, Any]])

Bases: object

Wrapper around a Mapping[str, str] where both the keys and values are to be interpolated.

mapping

to be evaluated

Type

Mapping[str, str]

eval(config: Mapping[str, Any], **additional_parameters)

Wrapper around a Mapping[str, str] that allows for both keys and values to be interpolated.

Parameters

• config – The user-provided configuration as specified by the source’s spec

• additional_parameters – Optional parameters used for interpolation

Returns

The interpolated string

mapping: Mapping[str, str]

parameters: dataclasses.InitVar[Mapping[str, Any]]

class airbyte_cdk.sources.declarative.interpolation.InterpolatedString(string: str, parameters: dataclasses.InitVar[Mapping[str, Any]], default: Optional[str] = None)

Bases: object

Wrapper around a raw string to be interpolated with the Jinja2 templating engine

string

The string to evalute

Type

str

default

The default value to return if the evaluation returns an empty string

Type

Optional[str]

parameters

Additional runtime parameters to be used for string interpolation

Type

Mapping[str, Any]

classmethod create(string_or_interpolated: Union[airbyte_cdk.sources.declarative.interpolation.interpolated_string.InterpolatedString, str], *, parameters: Mapping[str, Any])

Helper function to obtain an InterpolatedString from either a raw string or an InterpolatedString.

Parameters

• string_or_interpolated – either a raw string or an InterpolatedString.

• parameters – parameters propagated from parent component

Returns

InterpolatedString representing the input string.

default: Optional[str] = None

eval(config: Mapping[str, Any], **kwargs)

Interpolates the input string using the config and other optional arguments passed as parameter.

Parameters

• config – The user-provided configuration as specified by the source’s spec

• kwargs – Optional parameters used for interpolation

Returns

The interpolated string

parameters: dataclasses.InitVar[Mapping[str, Any]]

string: str