Pipeline API

Pipeline API provides building blocks for YAML pipeline.

Each stage and configuration object here represents a specific part of the YAML definition file.

Example pipeline stage with all possible keys. Actual usage of keys depends on used model kind.

- name: str
  kind: source | sink | transform
  log_level: DEBUG
  skip_if: 
  on_error: error | continue
  connection: Connection
    kind: Parquet | Deltatable | Rest | ...
    locator: str
    config: ConnectionConfiguration
    client: RestClient
    request: RestRequest
      endpoint: str
      method: RequestMethod
      errorhandler: RestErrorHandler
      auth: RestAuth
        method: header | oauth2 | ...
        values: StringDict
      query: dict | @json-magic
      body: dict | @json-magic
      response: RestResponse
        handler: 
            kind: ResponseHandlerTypes
            page_param: str
            page_size: str
            param_locator: ParameterDisposition
            total_records: str
            throttle: int | float 
        content_type: DataType
        locator: str
    fields: Fields
  show: int
  show_schema: bool
  query: > str
  stages: Stages
  throttle: int

all ¶

__all__ = [
    "Column",
    "Columns",
    "Stages",
    "Stage",
    "ConnectionConfiguration",
    "Variables",
    "FlowContext",
    "S3Config",
    "GSConfig",
    "ClickhouseConfig",
    "BigQueryConfig",
    "BucketConfig",
    "Connection",
    "PhysicalConnection",
    "EphemeralConnection",
    "ParquetConnection",
    "CSVConnection",
    "JSONConnection",
    "FileConnection",
    "BigQueryConnection",
    "ClickhouseConnection",
    "DeltatableConnection",
    "Dimension",
    "CustomConnection",
    "RestConnection",
    "SQLGenConnection",
    "OAuth2Provider",
]

BigQueryConfig ¶

Bases: BaseModel

Configuration for accessing BigQuery datasets.

credential_file ¶

credential_file: str | None = None

Path to service account JSON credentials for BigQuery access.

dataset ¶

dataset: str | None = None

BigQuery dataset name.

project ¶

project: str | None = None

GCP project ID containing the BigQuery dataset.

region ¶

region: str | None = None

BigQuery region (e.g., "us-central1").

BigQueryConnection ¶

Bases: PhysicalConnection, VersionedConnection

data_mode ¶

data_mode: str = 'error'

Data mode for write operation. For Deltatable valid options are:

append adds new data
overwrite replaces all data
error fails (table is read-only)

key ¶

key: list[str] | None = None

List of versioned fields

kind ¶

kind: Literal['BigQuery']

partition ¶

partition: list[str] | None = None

If set then delta table is partitioned using specified fields for faster reads

schema_mode ¶

schema_mode: str | None = None

Deltatable schema behaviour. If not set then write fails with error in case the data schema does not match existing table schema.

Schema evolution options:

merge adds new columns from data
overwrite adds new columns, drops missing

version ¶

version: str | None = None

Field for record version timestamp

BucketConfig ¶

Bases: BaseModel

bucket ¶

bucket: str | None = None

Bucket name eg. s3://my-bucket or local absolute path eg. /data/path

data_prefix ¶

data_prefix: str | None = None

Prefix for data files: s3://my-bucket/<prefix> or /data/path/<prefix>

locator_wildcard ¶

locator_wildcard: tuple | None = None

Regular expression and wildcard to modify locator. Useful in cases when you sink data to data-YYYY.parquet but want to read data-*.parquet. Provide tuple with patten as first element and replacement as second one. Example:

('-\d{4}-', '*')

This will create wildcard for data-YYYY-base.parquet as data*base.parquet.

Wildcard	Description
`*`	matches any number of any characters (including none)
`**`	matches any number of subdirectories (including none)
`[abc]`	matches one character given in the bracket
`[a-z]`	matches one character from the range given in the bracket

Wildcard is automatically applied in tap and show_schema operations.

region ¶

region: str | None = None

CSVConnection ¶

Bases: PhysicalConnection

ClickhouseConfig ¶

Bases: DatabaseConfig

Configuration for ClickHouse database, extending generic SQL config.

blocksize ¶

blocksize: int = 50000

Number of rows to process per block for batch operations.

ClickhouseConnection ¶

Bases: PhysicalConnection, VersionedConnection

key ¶

key: list[str] | None = None

List of versioned fields

kind ¶

kind: Literal['Clickhouse']

version ¶

version: str | None = None

Field for record version timestamp

Column ¶

Bases: BaseModel

Data column (equivalent to database column)

name ¶

name: str

Column name must follow rules set to SQL engine column names

type ¶

type: str

Any data type support by SQL engine

Columns ¶

Bases: RootModel[list[Column]]

Iterable list-like collection of Fields.

Connection ¶

Bases: BaseModel

config ¶

config: ConnectionConfiguration | None = None

Optional configuration for the current connection. If not present then global configuration will be used.

create_statement ¶

create_statement: str | None = None

Create statement for given table. Must be in requested dialect.

fields ¶

fields: Columns | None = None

If set then schema is used to generate source structure in case actual source does not provide data in which case generation of ephemeral view fails.

kind ¶

kind: str

Model type e.g. Deltatable, Bigquery, Clickhouse, Parquet, File Custom connections can be loaded from module.

params ¶

params: dict | None = PydanticField(default_factory=dict)

Any parameters that can be passed to connection.

show_schema ¶

show_schema: bool | None = None

If true then schema is automatically detected from the input data and logged '

ConnectionConfiguration ¶

Bases: BaseModel

Top-level container for all connection configurations in a pipeline.

Includes default configuration blocks for supported sources and sinks like: - Local file systems - S3 and GCS buckets - BigQuery datasets - ClickHouse databases

These can be customized per pipeline or extended with new sources.

Example

To support MySQL, you can define a new model like this:

class MySQLConfig(DatabaseConfig):
    pass  # You can add MySQL-specific fields here if needed

class ExtendedConnectionConfiguration(ConnectionConfiguration):
    mysql: MySQLConfig = PydanticField(default_factory=MySQLConfig)

This will allow you to specify MySQL connection settings in your YAML:

connection:
  kind: MySQL
  config:
    host: localhost
    port: 3306
    database: sales
    username: admin
    password: secret

bigquery ¶

bigquery: BigQueryConfig = PydanticField(
    default_factory=BigQueryConfig
)

BigQuery connection configuration.

clickhouse ¶

clickhouse: ClickhouseConfig = PydanticField(
    default_factory=ClickhouseConfig
)

Language model configuration.

gs ¶

gs: GSConfig = PydanticField(default_factory=GSConfig)

Google Cloud Storage configuration.

llm ¶

llm: LLMConfig = PydanticField(default_factory=LLMConfig)

Language model configuration.

local ¶

local: BucketConfig = PydanticField(
    default_factory=BucketConfig
)

Local file system configuration.

s3 ¶

s3: S3Config = PydanticField(default_factory=S3Config)

S3 cloud storage configuration.

CustomConnection ¶

Bases: PhysicalConnection

Custom connection provider. Custom connection may implement its own logic but must derive from base Connection class, and expose tap(), sink(), sql() and show_schema() even if they are no-op.

classname ¶

classname: str

Name of the class to load from the module

module ¶

module: str

Python module where the connection class is defined

params ¶

params: dict | None = PydanticField(default_factory=dict)

Free-form configuration parameters passed to the loaded class

load ¶

load() -> Type[Connection]

Dynamically load the connection class from the given module and member.

Returns:

Type	Description
`Type[Connection]`	A subclass of Connection.

Raises:

Type	Description
`ImportError`	If the module or member can't be imported.

Source code in ankaflow/models/connections.py

def load(self) -> t.Type[Connection]:
    """
    Dynamically load the connection class from the given module and member.

    Returns:
        A subclass of Connection.

    Raises:
        ImportError: If the module or member can't be imported.
    """
    mod = importlib.import_module(self.module)
    try:
        cls = getattr(mod, self.classname)
    except AttributeError as e:
        raise ImportError(
            f"Could not load '{self.classname}' from module '{self.module}'"
        ) from e
    # TODO: Refactor imports to facilitate this check early
    # if not issubclass(cls, Connection):
    #     raise TypeError(
    #         f"{cls.__name__} is not a subclass of Connection"
    #     )

    return cls

DeltatableConnection ¶

Bases: PhysicalConnection, VersionedConnection

data_mode ¶

data_mode: str = 'error'

Data mode for write operation. For Deltatable valid options are:

append adds new data
overwrite replaces all data
error fails (table is read-only)

kind ¶

kind: Literal['Deltatable']

optimize ¶

optimize: str | int | None = 1

Use with Deltatable and other engines whether to optimize after each sink operation. With larger tables this may be a lengthy synchronous operation.

Default value is optimize and vacuum with 7 day retention.

Deltatable¶

Values are optimize,vacuum,all,Literal[int]. If value is literal int is provided then parts older than number of days will be removed. Note that this will override default retention days.

String options vacuum,all are equivalent to 0.

Delta connection also supports raw SQL optimization. Pass statement as sql command:

OPTIMIZE Deltatable [COMPACT] [VACUUM] [AGE=<int>[d|h]] [DRY_RUN] [CLEANUP]

optimize deltatable → compact + vacuum with default 7 days

optimize deltatable compact → compact only

optimize deltatable vacuum age=36h dry_run → list files older than 36 hours, don't delete

optimize deltatable compact vacuum age=1 cleanup → compact, vacuum 1 day, then cleanup metadata

partition ¶

partition: list[str] | None = None

If set then delta table is partitioned using specified fields for faster reads

schema_mode ¶

schema_mode: str | None = None

Deltatable schema behaviour. If not set then write fails with error in case the data schema does not match existing table schema.

Schema evolution options:

merge adds new columns from data
overwrite adds new columns, drops missing

writer_features ¶

writer_features: Optional[List] | None = None

Any supported Delta-rs parameters can be passed to the writer. Example:

writer_features: [TimestampWithoutTimezone]

Dimension ¶

Bases: EphemeralConnection

load ¶

load() -> Type[Connection]

Dynamically load the connection class from the given module and member.

Returns:

Type	Description
`Type[Connection]`	A subclass of Connection.

Raises:

Type	Description
`ImportError`	If the module or member can't be imported.

Source code in ankaflow/models/connections.py

def load(self) -> t.Type[Connection]:
    """
    Dynamically load the connection class from the given module and member.

    Returns:
        A subclass of Connection.

    Raises:
        ImportError: If the module or member can't be imported.
    """
    mod = importlib.import_module(self.module)
    try:
        cls = getattr(mod, self.kind)
    except AttributeError as e:
        raise ImportError(
            f"Could not load '{self.kind}' from module '{self.module}'"
        ) from e

    return cls

EphemeralConnection ¶

Bases: Connection

FileConnection ¶

Bases: PhysicalConnection

GSConfig ¶

Bases: BucketConfig

Google Cloud Storage (GCS) configuration.

HMAC credentials allow reading/writing Parquet, JSON, and CSV.
Delta table writes require a full service account credential file.

credential_file ¶

credential_file: str | None = None

Path to GCP service account credential file (JSON). Required for certain write operations (e.g., Delta tables).

hmac_key ¶

hmac_key: str | None = None

GCS HMAC key ID for authenticated access.

hmac_secret ¶

hmac_secret: str | None = None

GCS HMAC secret for authenticated access.

JSONConnection ¶

Bases: PhysicalConnection

OAuth2Provider ¶

Bases: BaseModel

Implements standard OAuth2 authorization and token exchange flows.

Retry Policy: "The auth mechanism automatically retries transient errors (5xx, 429, and timeouts) up to 3 times before triggering on_refresh_fail."

access_token ¶

access_token: str | None = None

Current valid access token (Bearer). In the event token exchange fails this will be set to None immediately.

name ¶

name: str

Unique identifier used in YAML connection 'provider' field.

on_refresh_fail ¶

on_refresh_fail: RefreshFailCallable | None = None

Callback triggered when refresh fails (4xx/5xx or malformed response). (eg notifying UI or invalidating old tokens)

Signature: (name: str, error_context: dict) -> None

on_token_refresh ¶

on_token_refresh: RefreshSuccessCallable | None = None

Optional callback triggered on successful token update. (eg storing in secrets manager)

Signature: (name: str, data: dict) -> None

refresh_token ¶

refresh_token: str | None = None

Long-lived token used to obtain new access tokens.

subject_token ¶

subject_token: str | None = None

The session-specific identity token used for RFC 8693 exchanges.

ParquetConnection ¶

Bases: PhysicalConnection

PhysicalConnection ¶

Bases: Connection

locator ¶

locator: str

Table name or file name or URI, or other identifier required by the connection.

raw_dispatch ¶

raw_dispatch: bool | None = None

If True, sends the provided SQL query directly without altering locators if fully qualified, or adding FROM clauses. The query must be a valid SELECT statement. Useful for full control over complex queries.

In Deltatable and Parquet connections supports rewriting short locators as convenience.

RestConnection ¶

Bases: EphemeralConnection

Configuration for a REST-based data connection.

This model defines how to configure and execute REST API requests as part of a pipeline step. It includes the request definition, client behavior (e.g., retries, headers), and optional schema discovery.

client ¶

client: RestClientConfig

Configuration for the REST client (e.g., base URL, headers, auth).

fields ¶

fields: Columns | None = None

Optional schema definition used to validate or transform response data.

It is recommended to manually specify the schema after initial discovery. This ensures downstream pipeline stages remain stable, even when the remote API returns no results (e.g., due to no updates in an incremental fetch). Explicit schema prevents silent failures or schema drift in such cases.

kind ¶

kind: Literal['Rest']

Specifies the connection type as "Rest".

request ¶

request: Request

Request template specifying method, path, body, etc.

show_schema ¶

show_schema: bool | None = None

If True, the connector will attempt to infer or display the response schema automatically.

S3Config ¶

Bases: BucketConfig

S3-specific bucket configuration including optional authentication.

access_key_id ¶

access_key_id: str | None = None

AWS access key ID for private S3 access.

secret_access_key ¶

secret_access_key: str | None = None

AWS secret access key for private S3 access.

SQLGenConnection ¶

Bases: EphemeralConnection

SQLGen connection is intended for SQL code generation:

Prompt should instruct the model to generate SQL query and and executes resutling a VIEW in the internal database.

Example:

Stage 1: Name: ReadSomeParquetData

Stage 2: Name: CodeGen, query: Given SQL table `ReadSomeParquetData` generate SQL query to
    count number of rows in the table.

Inside stage 2 the following happens:

1. Prompt is sent to inferecne endpoint

2. Endpoint is expected to respond with valid SQL

3. Connection will execute a statement `CREATE OR REPLACE VIEW StageName AS <received_select_statement>`
    where statement in the exmaple is likely `SELECT COUNT() FROM ReadSomeParquetData`

.

kind ¶

kind: Literal['SQLGen']

Specifies the kind==SQLGen

variables ¶

variables: dict | None = None

Variables passed to Prompt. Prompt must be supplied in the query field of the Stage.

Prompt may contain Jinja2-style placeholders:

Example

Here's my name: {{name}}.

The connection will render the prompt template using variables

Stage ¶

Bases: BaseModel

connection ¶

connection: (
    RestConnection
    | VariableConnection
    | BigQueryConnection
    | DeltatableConnection
    | ParquetConnection
    | ClickhouseConnection
    | CustomConnection
    | JSONConnection
    | CSVConnection
    | FileConnection
    | SQLGenConnection
    | Dimension
    | None
) = Field(None, discriminator="kind")

Defines how the data is read from / written to the target.

Connection fields may contain templates and they will be recursively.

Special construct is JSON> which allows dynamically generating parameters as runtime:

- kind: source
  name: source_name
  connection:
    kind: File
    params: >
      JSON>{
        "key": "value",
        "dynkey": <<API.property>>,
      }

In the above app params are constructed as JSON string. It is possible to even construct parameter keys dynamically:

params: >
  JSON>
  {
    <% for number in [1,2,3] %>
    "key_<< number >>":<< number >><% if not loop.last %>,<% endif %>
    <% endfor %>
  }

Above example results the following:

params: >
  {
    "key_1": 1,
    "key_2": 2,
    "key_3": 3
  }

JSON> structure cannot contain nested JSON> structures, the entire string following the JSON> header must result a valid JSON.

Inbuilt connections include:

Deltatable (read)

If connection is Deltatable then query is required to narrow down data stored in the delta table. FROM clause must be Deltatable: - kind: source name: delta_tap connection: kind: Deltatable locator: delta_table query: > select * from Deltatable

Deltatable (write)

See also StorageOptions

The following example writes data from preceding stage to delta table, appending the data, partitions using part column, and optimizes and vacuums immediately without retention after write. - kind: sink name: delta_sink connection: kind: Deltatable locator: delta_table optimize: 0 data_mode: append partition: - part_field

Parquet (read/write)
JSON (read/write; NOTE: write operation generates newline-delimited JSON)
CSV (read/write)
Variable (read/write)
File (read)

File can be read from a connected filesystem (including s3). File name and file type must be specified in the pipeline context:

- `context.FileName`: file name relative to `file_prefix`
- `context.FileType`: file type, one of CSV, XLSX, JSON, HTML

Any file reader configuration must be passed in params.

Rest (bidirectional)

Rest connections consist of two parts: Client and Request. Client contains base URL and authentication (basic, digest, header and oauth2 are supported).

- name: TheRest kind: source connection: kind: Rest client: base_url: https://api.example.com auth: # Every request to given endpoint share the same authentication method: basic values: username: TheName password: ThePassword request: endpoint: /v1/some-path content_type: application/json method: post query: # Query parameters date: <<API.dt(None).isoformat()>> body: # JSON payload param: 1 response: content_type: application/json locator: "JMESPath.to.data"

Any custom source (API) can be used as long as available via module.

context ¶

context: FlowContext | None = None

@private Global context passed to given stage

explain ¶

explain: bool | None = None

If set to true then SQL query explanation will be logged.

fields ¶

fields: list[Column] | None = None

Explicitly defined output fields.

kind ¶

kind: str

Defines which action will be performed:

source
transform
sink
pipeline
sql

Sink¶

Sink reads output from previous stage and stores in specified location.

NOTE: If query is supplied with the stage then sink uses output of the query rather than preceding stage directly. Subsequent sink will use preceding stage. If supplied query must create either view or table with same name as current stage.

- name: my_sink
kind: sink
connection:
    kind: Variable
query: >
    CREATE VIEW my_sink AS
    SELECT
    1 as foo

Pipeline¶

If pipeline is preceded by any stage then the subpipeline will be executed as many times as there are rows in the previous stage output. This is useful if you want to run same pipeline with different parameters. Make sure the pipeline is preceded by source or transform producing required number of rows. If you need to run subpipeline only once there are two options:

Place it to the top
Preced with tranform producing single row only

Each row is then passed to subpipeline in a special variable.

Example pipeline iterating subpipeline 5 times:

kind: transform name: generate_5 # Generate 5 rows query: > select unnest as current_variable from unnest(generate_series(1,5)) show: 5
kind: pipeline name: looped_5x stages: - kind: transform name: inside_loop # In query we can reference the value passed from parent pipeline query: > select 'Currently running iteration: {API.look('loop_control.current_variable', variables)}' as value show: 5

locator ¶

locator: str | None = None

Currently unused: Name for the connection configuration: name, or URI.

log_level ¶

log_level: LogLevel | None = None

Set logging level. All stages after (including current) will log with specified level. Possible values: INFO (default), DEBUG, WARNING. Log level will be reset to INFO after each pipeline (including nested pipelines).

name ¶

name: str

Name of the stage, must be unique across all stages in the pipeline and conform to the rules: Must start with letter, may contain lowercase letters, number and underscores. Name is used to reference this stage by other subsequent stages.

on_error ¶

on_error: str = 'fail'

If set to 'continue' then pipeline will not fail. Subsequent stages referring to failed one must handle missing data.

query ¶

query: str | None = None

SQL Query or dictionary with custom source parameters. May contain {dynamic variables}.

show ¶

show: int = 0

If set to positive integer then given number of rows from this stage will get logged. If set to -1 then all rows will be loggged. Set to 0 to disable logging.

show_schema ¶

show_schema: bool | None = None

If True then schema is logged

skip_if ¶

skip_if: Any | None = None

Any value that can evaluated using bool(). or template string e.g. << True >>. When the expression evaluates to True then the stage is skipped.

stages ¶

stages: Stages | None = None

Used when kind is Flow

throttle ¶

throttle: int | float | None = None

If set to positive value then flow execution will be paused after the stage for given number of seconds.

Useful when dealing with rate limits or otherwise spreading the load over time.

Stages ¶

Bases: RootModel[list[Stage]]

A sequence of processing stages in a data pipeline.

Each Datablock in root represents one discrete step or transformation in an end-to-end business workflow (e.g., tap, transform, sink, validate).

Attributes:

Name	Type	Description
`root`	`List[Datablock]`	Ordered list of pipeline stages to execute.

enumerate_steps ¶

enumerate_steps() -> Iterator[tuple[int, Stage]]

Yield each stage along with its 0-based position.

Use this when you need both the stage and its index for logging, metrics, or conditional branching.

Returns:

Type	Description
`Iterator[tuple[int, Stage]]`	Iterator[Tuple[int, Datablock]]: Pairs of (index, stage).

Source code in ankaflow/models/core.py

def enumerate_steps(self) -> t.Iterator[tuple[int, Stage]]:
    """Yield each stage along with its 0-based position.

    Use this when you need both the stage and its index for logging,
    metrics, or conditional branching.

    Returns:
        Iterator[Tuple[int, Datablock]]: Pairs of (index, stage).
    """
    return enumerate(self.root)

load ¶

load(source: str | Path | IO[str] | Loadable) -> Stages

Load a pipeline from YAML (path, YAML-string, file-like or Loadable).

Parameters:

Name	Type	Description	Default
`source`	`str \| Path \| IO[str] \| Loadable`	Path to a .yaml file Raw YAML content File-like object returning YAML Any object with a `.load()` method returning Python data	required

Returns:

Name	Type	Description
`Stages`	`Stages`	a validated `Stages` instance.

Source code in ankaflow/models/core.py

@classmethod
def load(
    cls,
    source: t.Union[str, Path, t.IO[str], Loadable],
) -> "Stages":
    """Load a pipeline from YAML (path, YAML-string, file-like or Loadable).

    Args:
        source (str | Path | IO[str] | Loadable):
            - Path to a .yaml file
            - Raw YAML content
            - File-like object returning YAML
            - Any object with a `.load()` method returning Python data

    Returns:
        Stages: a validated `Stages` instance.
    """
    # 1) If it’s a loader-object, pull Python data directly
    if isinstance(source, Loadable):
        data = source.load()

    else:
        # 2) Read text for YAML parsing:
        if hasattr(source, "read"):
            text = t.cast(t.IO[str], source).read()
        else:
            text = str(source)

        # 3) First, try parsing as raw YAML
        try:
            data = yaml.safe_load(text)
        except yaml.YAMLError:
            data = None

        # 4) Only if that parse returned a `str` we treat it as a filename
        if isinstance(data, str):
            try:
                text = Path(data).read_text()
                data = yaml.safe_load(text)
            except (OSError, yaml.YAMLError) as e:
                raise ValueError(
                    f"Could not interpret {data!r} as YAML or file path"
                ) from e

    # 5) Validate final shape
    if not isinstance(data, list):
        raise ValueError(
            f"Expected a list of pipeline stages, got {type(data).__name__}"
        )

    # 5) Finally, validate into our model
    return cls.model_validate(data)

steps ¶

steps() -> Iterator[Stage]

Yield each stage in execution order.

Returns:

Type	Description
`Iterator[Stage]`	Iterator[Datablock]: An iterator over the stages,
`Iterator[Stage]`	from first to last.

Source code in ankaflow/models/core.py

def steps(self) -> t.Iterator[Stage]:
    """Yield each stage in execution order.

    Returns:
        Iterator[Datablock]: An iterator over the stages,
        from first to last.
    """
    return iter(self.root)

components ¶

Column ¶

Bases: BaseModel

Data column (equivalent to database column)

name ¶

name: str

Column name must follow rules set to SQL engine column names

type ¶

type: str

Any data type support by SQL engine

Columns ¶

Bases: RootModel[list[Column]]

Iterable list-like collection of Fields.

configs ¶

BigQueryConfig ¶

Bases: BaseModel

Configuration for accessing BigQuery datasets.

credential_file ¶

credential_file: str | None = None

Path to service account JSON credentials for BigQuery access.

dataset ¶

dataset: str | None = None

BigQuery dataset name.

project ¶

project: str | None = None

GCP project ID containing the BigQuery dataset.

region ¶

region: str | None = None

BigQuery region (e.g., "us-central1").

BucketConfig ¶

Bases: BaseModel

bucket ¶

bucket: str | None = None

Bucket name eg. s3://my-bucket or local absolute path eg. /data/path

data_prefix ¶

data_prefix: str | None = None

Prefix for data files: s3://my-bucket/<prefix> or /data/path/<prefix>

locator_wildcard ¶

locator_wildcard: tuple | None = None

Regular expression and wildcard to modify locator. Useful in cases when you sink data to data-YYYY.parquet but want to read data-*.parquet. Provide tuple with patten as first element and replacement as second one. Example:

('-\d{4}-', '*')

This will create wildcard for data-YYYY-base.parquet as data*base.parquet.

Wildcard	Description
`*`	matches any number of any characters (including none)
`**`	matches any number of subdirectories (including none)
`[abc]`	matches one character given in the bracket
`[a-z]`	matches one character from the range given in the bracket

Wildcard is automatically applied in tap and show_schema operations.

region ¶

region: str | None = None

ClickhouseConfig ¶

Bases: DatabaseConfig

Configuration for ClickHouse database, extending generic SQL config.

blocksize ¶

blocksize: int = 50000

Number of rows to process per block for batch operations.

ConnectionConfiguration ¶

Bases: BaseModel

Top-level container for all connection configurations in a pipeline.

Includes default configuration blocks for supported sources and sinks like: - Local file systems - S3 and GCS buckets - BigQuery datasets - ClickHouse databases

These can be customized per pipeline or extended with new sources.

Example

To support MySQL, you can define a new model like this:

class MySQLConfig(DatabaseConfig):
    pass  # You can add MySQL-specific fields here if needed

class ExtendedConnectionConfiguration(ConnectionConfiguration):
    mysql: MySQLConfig = PydanticField(default_factory=MySQLConfig)

This will allow you to specify MySQL connection settings in your YAML:

connection:
  kind: MySQL
  config:
    host: localhost
    port: 3306
    database: sales
    username: admin
    password: secret

bigquery ¶

bigquery: BigQueryConfig = PydanticField(
    default_factory=BigQueryConfig
)

BigQuery connection configuration.

clickhouse ¶

clickhouse: ClickhouseConfig = PydanticField(
    default_factory=ClickhouseConfig
)

Language model configuration.

gs ¶

gs: GSConfig = PydanticField(default_factory=GSConfig)

Google Cloud Storage configuration.

llm ¶

llm: LLMConfig = PydanticField(default_factory=LLMConfig)

Language model configuration.

local ¶

local: BucketConfig = PydanticField(
    default_factory=BucketConfig
)

Local file system configuration.

s3 ¶

s3: S3Config = PydanticField(default_factory=S3Config)

S3 cloud storage configuration.

DatabaseConfig ¶

Bases: BaseModel

Base class for SQL database connection configurations.

cluster ¶

cluster: str | None = None

Optional cluster name or identifier (used in ClickHouse and other distributed systems).

database ¶

database: str | None = None

Database name.

host ¶

host: str | None = None

Hostname or IP address of the database server.

password ¶

password: str | None = None

Password for authentication.

port ¶

port: int | str | None = None

Database port.

username ¶

username: str | None = None

Username for authentication.

GSConfig ¶

Bases: BucketConfig

Google Cloud Storage (GCS) configuration.

HMAC credentials allow reading/writing Parquet, JSON, and CSV.
Delta table writes require a full service account credential file.

credential_file ¶

credential_file: str | None = None

Path to GCP service account credential file (JSON). Required for certain write operations (e.g., Delta tables).

hmac_key ¶

hmac_key: str | None = None

GCS HMAC key ID for authenticated access.

hmac_secret ¶

hmac_secret: str | None = None

GCS HMAC secret for authenticated access.

S3Config ¶

Bases: BucketConfig

S3-specific bucket configuration including optional authentication.

access_key_id ¶

access_key_id: str | None = None

AWS access key ID for private S3 access.

secret_access_key ¶

secret_access_key: str | None = None

AWS secret access key for private S3 access.

connections ¶

BigQueryConnection ¶

Bases: PhysicalConnection, VersionedConnection

data_mode ¶

data_mode: str = 'error'

Data mode for write operation. For Deltatable valid options are:

append adds new data
overwrite replaces all data
error fails (table is read-only)

key ¶

key: list[str] | None = None

List of versioned fields

kind ¶

kind: Literal['BigQuery']

partition ¶

partition: list[str] | None = None

If set then delta table is partitioned using specified fields for faster reads

schema_mode ¶

schema_mode: str | None = None

Deltatable schema behaviour. If not set then write fails with error in case the data schema does not match existing table schema.

Schema evolution options:

merge adds new columns from data
overwrite adds new columns, drops missing

version ¶

version: str | None = None

Field for record version timestamp

CSVConnection ¶

Bases: PhysicalConnection

ClickhouseConnection ¶

Bases: PhysicalConnection, VersionedConnection

key ¶

key: list[str] | None = None

List of versioned fields

kind ¶

kind: Literal['Clickhouse']

version ¶

version: str | None = None

Field for record version timestamp

Connection ¶

Bases: BaseModel

config ¶

config: ConnectionConfiguration | None = None

Optional configuration for the current connection. If not present then global configuration will be used.

create_statement ¶

create_statement: str | None = None

Create statement for given table. Must be in requested dialect.

fields ¶

fields: Columns | None = None

If set then schema is used to generate source structure in case actual source does not provide data in which case generation of ephemeral view fails.

kind ¶

kind: str

Model type e.g. Deltatable, Bigquery, Clickhouse, Parquet, File Custom connections can be loaded from module.

params ¶

params: dict | None = PydanticField(default_factory=dict)

Any parameters that can be passed to connection.

show_schema ¶

show_schema: bool | None = None

If true then schema is automatically detected from the input data and logged '

CustomConnection ¶

Bases: PhysicalConnection

Custom connection provider. Custom connection may implement its own logic but must derive from base Connection class, and expose tap(), sink(), sql() and show_schema() even if they are no-op.

classname ¶

classname: str

Name of the class to load from the module

module ¶

module: str

Python module where the connection class is defined

params ¶

params: dict | None = PydanticField(default_factory=dict)

Free-form configuration parameters passed to the loaded class

load ¶

load() -> Type[Connection]

Dynamically load the connection class from the given module and member.

Returns:

Type	Description
`Type[Connection]`	A subclass of Connection.

Raises:

Type	Description
`ImportError`	If the module or member can't be imported.

Source code in ankaflow/models/connections.py

def load(self) -> t.Type[Connection]:
    """
    Dynamically load the connection class from the given module and member.

    Returns:
        A subclass of Connection.

    Raises:
        ImportError: If the module or member can't be imported.
    """
    mod = importlib.import_module(self.module)
    try:
        cls = getattr(mod, self.classname)
    except AttributeError as e:
        raise ImportError(
            f"Could not load '{self.classname}' from module '{self.module}'"
        ) from e
    # TODO: Refactor imports to facilitate this check early
    # if not issubclass(cls, Connection):
    #     raise TypeError(
    #         f"{cls.__name__} is not a subclass of Connection"
    #     )

    return cls

DeltatableConnection ¶

Bases: PhysicalConnection, VersionedConnection

data_mode ¶

data_mode: str = 'error'

Data mode for write operation. For Deltatable valid options are:

append adds new data
overwrite replaces all data
error fails (table is read-only)

kind ¶

kind: Literal['Deltatable']

optimize ¶

optimize: str | int | None = 1

Use with Deltatable and other engines whether to optimize after each sink operation. With larger tables this may be a lengthy synchronous operation.

Default value is optimize and vacuum with 7 day retention.

Deltatable¶

Values are optimize,vacuum,all,Literal[int]. If value is literal int is provided then parts older than number of days will be removed. Note that this will override default retention days.

String options vacuum,all are equivalent to 0.

Delta connection also supports raw SQL optimization. Pass statement as sql command:

OPTIMIZE Deltatable [COMPACT] [VACUUM] [AGE=<int>[d|h]] [DRY_RUN] [CLEANUP]

optimize deltatable → compact + vacuum with default 7 days

optimize deltatable compact → compact only

optimize deltatable vacuum age=36h dry_run → list files older than 36 hours, don't delete

optimize deltatable compact vacuum age=1 cleanup → compact, vacuum 1 day, then cleanup metadata

partition ¶

partition: list[str] | None = None

If set then delta table is partitioned using specified fields for faster reads

schema_mode ¶

schema_mode: str | None = None

Deltatable schema behaviour. If not set then write fails with error in case the data schema does not match existing table schema.

Schema evolution options:

merge adds new columns from data
overwrite adds new columns, drops missing

writer_features ¶

writer_features: Optional[List] | None = None

Any supported Delta-rs parameters can be passed to the writer. Example:

writer_features: [TimestampWithoutTimezone]

Dimension ¶

Bases: EphemeralConnection

load ¶

load() -> Type[Connection]

Dynamically load the connection class from the given module and member.

Returns:

Type	Description
`Type[Connection]`	A subclass of Connection.

Raises:

Type	Description
`ImportError`	If the module or member can't be imported.

Source code in ankaflow/models/connections.py

def load(self) -> t.Type[Connection]:
    """
    Dynamically load the connection class from the given module and member.

    Returns:
        A subclass of Connection.

    Raises:
        ImportError: If the module or member can't be imported.
    """
    mod = importlib.import_module(self.module)
    try:
        cls = getattr(mod, self.kind)
    except AttributeError as e:
        raise ImportError(
            f"Could not load '{self.kind}' from module '{self.module}'"
        ) from e

    return cls

locator ¶

locator: str

Table name or file name or URI, or other identifier required by the connection.

raw_dispatch ¶

raw_dispatch: bool | None = None

If True, sends the provided SQL query directly without altering locators if fully qualified, or adding FROM clauses. The query must be a valid SELECT statement. Useful for full control over complex queries.

In Deltatable and Parquet connections supports rewriting short locators as convenience.

RestConnection ¶

Bases: EphemeralConnection

Configuration for a REST-based data connection.

This model defines how to configure and execute REST API requests as part of a pipeline step. It includes the request definition, client behavior (e.g., retries, headers), and optional schema discovery.

client ¶

client: RestClientConfig

Configuration for the REST client (e.g., base URL, headers, auth).

fields ¶

fields: Columns | None = None

Optional schema definition used to validate or transform response data.

It is recommended to manually specify the schema after initial discovery. This ensures downstream pipeline stages remain stable, even when the remote API returns no results (e.g., due to no updates in an incremental fetch). Explicit schema prevents silent failures or schema drift in such cases.

kind ¶

kind: Literal['Rest']

Specifies the connection type as "Rest".

request ¶

request: Request

Request template specifying method, path, body, etc.

show_schema ¶

show_schema: bool | None = None

If True, the connector will attempt to infer or display the response schema automatically.

SQLGenConnection ¶

Bases: EphemeralConnection

SQLGen connection is intended for SQL code generation:

Prompt should instruct the model to generate SQL query and and executes resutling a VIEW in the internal database.

Example:

Stage 1: Name: ReadSomeParquetData

Stage 2: Name: CodeGen, query: Given SQL table `ReadSomeParquetData` generate SQL query to
    count number of rows in the table.

Inside stage 2 the following happens:

1. Prompt is sent to inferecne endpoint

2. Endpoint is expected to respond with valid SQL

3. Connection will execute a statement `CREATE OR REPLACE VIEW StageName AS <received_select_statement>`
    where statement in the exmaple is likely `SELECT COUNT() FROM ReadSomeParquetData`

.

kind ¶

kind: Literal['SQLGen']

Specifies the kind==SQLGen

variables ¶

variables: dict | None = None

Variables passed to Prompt. Prompt must be supplied in the query field of the Stage.

Prompt may contain Jinja2-style placeholders:

Example

Here's my name: {{name}}.

The connection will render the prompt template using variables

VariableConnection ¶

Bases: PhysicalConnection

VersionedConnection ¶

Bases: BaseModel

key ¶

key: list[str] | None = None

List of versioned fields

version ¶

version: str | None = None

Field for record version timestamp

core ¶

Stage ¶

Bases: BaseModel

connection ¶

connection: (
    RestConnection
    | VariableConnection
    | BigQueryConnection
    | DeltatableConnection
    | ParquetConnection
    | ClickhouseConnection
    | CustomConnection
    | JSONConnection
    | CSVConnection
    | FileConnection
    | SQLGenConnection
    | Dimension
    | None
) = Field(None, discriminator="kind")

Defines how the data is read from / written to the target.

Connection fields may contain templates and they will be recursively.

Special construct is JSON> which allows dynamically generating parameters as runtime:

- kind: source
  name: source_name
  connection:
    kind: File
    params: >
      JSON>{
        "key": "value",
        "dynkey": <<API.property>>,
      }

In the above app params are constructed as JSON string. It is possible to even construct parameter keys dynamically:

params: >
  JSON>
  {
    <% for number in [1,2,3] %>
    "key_<< number >>":<< number >><% if not loop.last %>,<% endif %>
    <% endfor %>
  }

Above example results the following:

params: >
  {
    "key_1": 1,
    "key_2": 2,
    "key_3": 3
  }

JSON> structure cannot contain nested JSON> structures, the entire string following the JSON> header must result a valid JSON.

Inbuilt connections include:

Deltatable (read)

If connection is Deltatable then query is required to narrow down data stored in the delta table. FROM clause must be Deltatable: - kind: source name: delta_tap connection: kind: Deltatable locator: delta_table query: > select * from Deltatable

Deltatable (write)

See also StorageOptions

The following example writes data from preceding stage to delta table, appending the data, partitions using part column, and optimizes and vacuums immediately without retention after write. - kind: sink name: delta_sink connection: kind: Deltatable locator: delta_table optimize: 0 data_mode: append partition: - part_field

Parquet (read/write)
JSON (read/write; NOTE: write operation generates newline-delimited JSON)
CSV (read/write)
Variable (read/write)
File (read)

File can be read from a connected filesystem (including s3). File name and file type must be specified in the pipeline context:

- `context.FileName`: file name relative to `file_prefix`
- `context.FileType`: file type, one of CSV, XLSX, JSON, HTML

Any file reader configuration must be passed in params.

Rest (bidirectional)

Rest connections consist of two parts: Client and Request. Client contains base URL and authentication (basic, digest, header and oauth2 are supported).

- name: TheRest kind: source connection: kind: Rest client: base_url: https://api.example.com auth: # Every request to given endpoint share the same authentication method: basic values: username: TheName password: ThePassword request: endpoint: /v1/some-path content_type: application/json method: post query: # Query parameters date: <<API.dt(None).isoformat()>> body: # JSON payload param: 1 response: content_type: application/json locator: "JMESPath.to.data"

Any custom source (API) can be used as long as available via module.

context ¶

context: FlowContext | None = None

@private Global context passed to given stage

explain ¶

explain: bool | None = None

If set to true then SQL query explanation will be logged.

fields ¶

fields: list[Column] | None = None

Explicitly defined output fields.

kind ¶

kind: str

Defines which action will be performed:

source
transform
sink
pipeline
sql

Sink¶

Sink reads output from previous stage and stores in specified location.

NOTE: If query is supplied with the stage then sink uses output of the query rather than preceding stage directly. Subsequent sink will use preceding stage. If supplied query must create either view or table with same name as current stage.

- name: my_sink
kind: sink
connection:
    kind: Variable
query: >
    CREATE VIEW my_sink AS
    SELECT
    1 as foo

Pipeline¶

If pipeline is preceded by any stage then the subpipeline will be executed as many times as there are rows in the previous stage output. This is useful if you want to run same pipeline with different parameters. Make sure the pipeline is preceded by source or transform producing required number of rows. If you need to run subpipeline only once there are two options:

Place it to the top
Preced with tranform producing single row only

Each row is then passed to subpipeline in a special variable.

Example pipeline iterating subpipeline 5 times:

kind: transform name: generate_5 # Generate 5 rows query: > select unnest as current_variable from unnest(generate_series(1,5)) show: 5
kind: pipeline name: looped_5x stages: - kind: transform name: inside_loop # In query we can reference the value passed from parent pipeline query: > select 'Currently running iteration: {API.look('loop_control.current_variable', variables)}' as value show: 5

locator ¶

locator: str | None = None

Currently unused: Name for the connection configuration: name, or URI.

log_level ¶

log_level: LogLevel | None = None

Set logging level. All stages after (including current) will log with specified level. Possible values: INFO (default), DEBUG, WARNING. Log level will be reset to INFO after each pipeline (including nested pipelines).

name ¶

name: str

Name of the stage, must be unique across all stages in the pipeline and conform to the rules: Must start with letter, may contain lowercase letters, number and underscores. Name is used to reference this stage by other subsequent stages.

on_error ¶

on_error: str = 'fail'

If set to 'continue' then pipeline will not fail. Subsequent stages referring to failed one must handle missing data.

query ¶

query: str | None = None

SQL Query or dictionary with custom source parameters. May contain {dynamic variables}.

show ¶

show: int = 0

If set to positive integer then given number of rows from this stage will get logged. If set to -1 then all rows will be loggged. Set to 0 to disable logging.

show_schema ¶

show_schema: bool | None = None

If True then schema is logged

skip_if ¶

skip_if: Any | None = None

Any value that can evaluated using bool(). or template string e.g. << True >>. When the expression evaluates to True then the stage is skipped.

stages ¶

stages: Stages | None = None

Used when kind is Flow

throttle ¶

throttle: int | float | None = None

If set to positive value then flow execution will be paused after the stage for given number of seconds.

Useful when dealing with rate limits or otherwise spreading the load over time.

Stages ¶

Bases: RootModel[list[Stage]]

A sequence of processing stages in a data pipeline.

Each Datablock in root represents one discrete step or transformation in an end-to-end business workflow (e.g., tap, transform, sink, validate).

Attributes:

Name	Type	Description
`root`	`List[Datablock]`	Ordered list of pipeline stages to execute.

enumerate_steps ¶

enumerate_steps() -> Iterator[tuple[int, Stage]]

Yield each stage along with its 0-based position.

Use this when you need both the stage and its index for logging, metrics, or conditional branching.

Returns:

Type	Description
`Iterator[tuple[int, Stage]]`	Iterator[Tuple[int, Datablock]]: Pairs of (index, stage).

Source code in ankaflow/models/core.py

def enumerate_steps(self) -> t.Iterator[tuple[int, Stage]]:
    """Yield each stage along with its 0-based position.

    Use this when you need both the stage and its index for logging,
    metrics, or conditional branching.

    Returns:
        Iterator[Tuple[int, Datablock]]: Pairs of (index, stage).
    """
    return enumerate(self.root)

load ¶

load(source: str | Path | IO[str] | Loadable) -> Stages

Load a pipeline from YAML (path, YAML-string, file-like or Loadable).

Parameters:

Name	Type	Description	Default
`source`	`str \| Path \| IO[str] \| Loadable`	Path to a .yaml file Raw YAML content File-like object returning YAML Any object with a `.load()` method returning Python data	required

Returns:

Name	Type	Description
`Stages`	`Stages`	a validated `Stages` instance.

Source code in ankaflow/models/core.py

@classmethod
def load(
    cls,
    source: t.Union[str, Path, t.IO[str], Loadable],
) -> "Stages":
    """Load a pipeline from YAML (path, YAML-string, file-like or Loadable).

    Args:
        source (str | Path | IO[str] | Loadable):
            - Path to a .yaml file
            - Raw YAML content
            - File-like object returning YAML
            - Any object with a `.load()` method returning Python data

    Returns:
        Stages: a validated `Stages` instance.
    """
    # 1) If it’s a loader-object, pull Python data directly
    if isinstance(source, Loadable):
        data = source.load()

    else:
        # 2) Read text for YAML parsing:
        if hasattr(source, "read"):
            text = t.cast(t.IO[str], source).read()
        else:
            text = str(source)

        # 3) First, try parsing as raw YAML
        try:
            data = yaml.safe_load(text)
        except yaml.YAMLError:
            data = None

        # 4) Only if that parse returned a `str` we treat it as a filename
        if isinstance(data, str):
            try:
                text = Path(data).read_text()
                data = yaml.safe_load(text)
            except (OSError, yaml.YAMLError) as e:
                raise ValueError(
                    f"Could not interpret {data!r} as YAML or file path"
                ) from e

    # 5) Validate final shape
    if not isinstance(data, list):
        raise ValueError(
            f"Expected a list of pipeline stages, got {type(data).__name__}"
        )

    # 5) Finally, validate into our model
    return cls.model_validate(data)

steps ¶

steps() -> Iterator[Stage]

Yield each stage in execution order.

Returns:

Type	Description
`Iterator[Stage]`	Iterator[Datablock]: An iterator over the stages,
`Iterator[Stage]`	from first to last.

Source code in ankaflow/models/core.py

def steps(self) -> t.Iterator[Stage]:
    """Yield each stage in execution order.

    Returns:
        Iterator[Datablock]: An iterator over the stages,
        from first to last.
    """
    return iter(self.root)

enums ¶

AuthType ¶

Bases: Enum

ContentType ¶

Bases: Enum

DataType ¶

Bases: Enum

LogLevel ¶

Bases: Enum

ModelType ¶

Bases: Enum

ParameterDisposition ¶

Bases: Enum

RequestMethod ¶

Bases: Enum

llm ¶

LLMConfig ¶

Bases: BaseModel

Confuration for Language Model

kind ¶

kind: LLMKind = LLMKind.MOCK

Language model provider: mock(default)|openai

model ¶

model: str | None = None

Language model, if not set then uses default

proxy ¶

proxy: LLMProxy | None = None

Reverse proxy used to connect to provider (optional)

LLMKind ¶

Bases: str, Enum

Supported LLM protocol backends.

MOCK ¶

MOCK = 'mock'

OPENAI ¶

OPENAI = 'openai'

LLMProxy ¶

Bases: BaseModel

Configuration for proxy-based LLM usage.

client ¶

client: RestClientConfig

request ¶

request: Request

rest ¶

BasicHandler ¶

Bases: BaseModel

A no-op response handler used when no special processing (like pagination or transformation) is required.

Typically used for single-response REST endpoints where the entire payload is returned in one request.

Attributes:

Name	Type	Description
`kind`	`Literal`	Specifies the handler type as BASIC.

kind ¶

kind: Literal[BASIC]

Specifies the handler type as Basic.

CursorPaginator ¶

Bases: BaseModel

A response handler for cursor-based paginated REST and GraphQL APIs.

Generates repeated requests by extracting a cursor token or URL from the response and injecting it into subsequent requests until an empty/null cursor or end-sentinel is met.

cursor_disposition ¶

cursor_disposition: CursorRequestDisposition = (
    CursorRequestDisposition.QUERY
)

Where to put the extracted cursor in the next request (QUERY, BODY, HEADER).

cursor_locator ¶

cursor_locator: CursorResponseLocator = (
    CursorResponseLocator.BODY
)

Where and how to locate the cursor in the response.

cursor_name ¶

cursor_name: str | None = None

Parameter name or path in the next request (e.g., 'cursor', 'start_after'). Supports dot-notation paths for nested dicts when cursor_disposition='body' (e.g., 'variables.cursor').

cursor_param ¶

cursor_param: str | None = None

Field or path name to extract from response:

JMESPath expression if cursor_locator='body' (e.g., 'nextCursor', 'data.orders.pageInfo.endCursor')
Header name if cursor_locator='header' (e.g., 'X-Next-Cursor')
Query parameter name if cursor_locator='rfc_token'
Ignored if cursor_locator='rfc_url'

stop_value ¶

stop_value: Any | None = None

Optional explicit value signalling end of pagination (e.g., 'END').

throttle ¶

throttle: int | float | None = None

Optional pause duration in seconds between page requests.

CursorRequestDisposition ¶

Bases: str, Enum

Specifies where and how to inject the pagination cursor into the next request.

BODY ¶

BODY = 'body'

Inject cursor into the JSON request body. Supports dot-notation paths for nested fields (e.g., 'variables.cursor').

QUERY ¶

QUERY = 'query'

Inject cursor as a URL query parameter (e.g., ?cursor=xyz123).

CursorResponseLocator ¶

Bases: str, Enum

Specifies where and how to extract the cursor from an incoming response.

BODY ¶

BODY = 'body'

Extract cursor from JSON response body using a JMESPath expression.

HEADER ¶

HEADER = 'header'

Read cursor directly from an HTTP response header.

RFC_TOKEN ¶

RFC_TOKEN = 'rfc_token'

Extract a named query parameter from an RFC 8288 Link header.

RFC_URL ¶

RFC_URL = 'rfc_url'

Extract full next URL/path from an RFC 8288 Link header.

OAuth2Config ¶

Bases: BaseModel

extra_params ¶

extra_params: dict = PydanticField(default_factory=dict)

Arbitrary key-value pairs for vendor-specific extensions (e.g., {'resource': 'https://analysis.windows.net/powerbi/api'}).

grant_type ¶

grant_type: str = (
    "urn:ietf:params:oauth:grant-type:token-exchange"
)

The grant type for the credential exchange (default: Token Exchange).

requested_token_type ¶

requested_token_type: str = (
    "urn:ietf:params:oauth:token-type:access_token"
)

The type of security token being requested from the provider.

subject_token ¶

subject_token: str | None = None

A security token that represents the identity of the party on whose behalf the request is being made.

subject_token_type ¶

subject_token_type: str = (
    "urn:ietf:params:oauth:token-type:id_token"
)

The identifier for the type of security token being provided.

OAuth2Provider ¶

Bases: BaseModel

Implements standard OAuth2 authorization and token exchange flows.

Retry Policy: "The auth mechanism automatically retries transient errors (5xx, 429, and timeouts) up to 3 times before triggering on_refresh_fail."

access_token ¶

access_token: str | None = None

Current valid access token (Bearer). In the event token exchange fails this will be set to None immediately.

name ¶

name: str

Unique identifier used in YAML connection 'provider' field.

on_refresh_fail ¶

on_refresh_fail: RefreshFailCallable | None = None

Callback triggered when refresh fails (4xx/5xx or malformed response). (eg notifying UI or invalidating old tokens)

Signature: (name: str, error_context: dict) -> None

on_token_refresh ¶

on_token_refresh: RefreshSuccessCallable | None = None

Optional callback triggered on successful token update. (eg storing in secrets manager)

Signature: (name: str, data: dict) -> None

refresh_token ¶

refresh_token: str | None = None

Long-lived token used to obtain new access tokens.

subject_token ¶

subject_token: str | None = None

The session-specific identity token used for RFC 8693 exchanges.

Paginator ¶

Bases: BaseModel

A response handler for paginated REST APIs.

This handler generates repeated requests by incrementing a page-related parameter until no more data is available. The stopping condition is usually inferred from the number of records in the response being less than page_size, or from a total record count field.

increment ¶

increment: int

Page parameter increment. Original request configuration should include initial value e.g. page_no=1

kind ¶

kind: Literal[PAGINATOR]

Specifies the handler type as Paginator.

page_param ¶

page_param: str

Page parameter in the request (query or body) This will be incremented from request to request

page_size ¶

page_size: int

Page size should be explicitly defined. If response contains less records it is considered to be last page

param_locator ¶

param_locator: ParameterDisposition

Define where the parameter is located: body or query

throttle ¶

throttle: int | float | None = None

If set to positive value then each page request is throttled given number of seconds.

Useful when dealing with rate limits or otherwise spreading the load over time.

total_records ¶

total_records: str | None = None

JMESPath to total records count in the response.

Request ¶

Bases: BaseModel

body ¶

body: str | dict | None = None

Request body parameters.

This field accepts either: - A Python dict representing a direct key-value mapping, or - A Jinja-templated JSON string with magic @json prefix, e.g.: @json{"parameter": "value"}

The template will be rendered using the following custom delimiters: - << ... >> for variable interpolation - <% ... %> for logic/control flow (e.g., for-loops) - <# ... #> for inline comments

The template will be rendered before being parsed into a valid JSON object. This allows the use of dynamic expressions, filters, and control flow such as loops.

Example with looping¶

Given:

variables = {
    "MyList": [
        {"id": 1, "value": 10},
        {"id": 2, "value": 20}
    ]
}

You can generate a dynamic body with:

body: >
@json[
    <% for row in API.look("MyTable", variables) %>
        { "id": << row.id >>, "value": << row.value >> }<% if not loop.last %>,<% endif %>
    <% endfor %>
]

This will render to a proper JSON list:

[
    { "id": 1, "value": 10 },
    { "id": 2, "value": 20 }
]

Notes: - When using @json, the entire string is rendered as a Jinja template and then parsed with json.loads(). - Nested @json blocks are not supported. - Newlines and whitespace are automatically collapsed during rendering.

content_type ¶

content_type: ContentType = ContentType.JSON

Request content type

endpoint ¶

endpoint: str

Request endpoint e.g. get/data under base url:

Example https://api.example.com/v1 + get/data

errorhandler ¶

errorhandler: RestErrorHandler = PydanticField(
    default_factory=RestErrorHandler
)

Custom error handler e.g. for searching conditions in response or custom status codes

initial_backoff ¶

initial_backoff: float = 0.5

Initial backoff time in seconds. Will be multiplied exponentially for subsequent retries (2^n).

max_retries ¶

max_retries: int = 0

Maximum number of retries on transport errors. Default is 0 (no retry).

method ¶

method: RequestMethod

Request method e.g. post,get,put

query ¶

query: dict = {}

Query parameters. Parameters may contain template variables.

response ¶

response: RestResponse

Response handling configuration

ResponseHandlerTypes ¶

RestAuth ¶

Bases: BaseModel

Authenctication configuration for Rest connection.

NOTE: Not all authentication methods may not work in browser due to limitations in the network API.

method ¶

method: AuthType

Specifies authentiation type.

oauth_header ¶

oauth_header: str | None = None

Some providers use different header pattern than Authorization-Bearer. If set then auth mechanism will user oauth_header=token

provider ¶

provider: str | OAuth2Provider | None = None

Either a string (lookup key for the keyring) or an inline OAuth2Provider definition.

values ¶

values: StringDict | dict = PydanticField(
    default_factory=dict
)

Mapping of parameter names and values.

{ 'X-Auth-Token': '' }

coerce_to_stringdict ¶

coerce_to_stringdict(v)

Rest header values must be strings. This convenience validator automatically converts regiular dictionary to StringDict.

Source code in ankaflow/models/rest.py

@field_validator("values", mode="before")
@classmethod
def coerce_to_stringdict(cls, v):
    """Rest header values must be strings.
    This convenience validator  automatically
    converts regiular dictionary to StringDict.
    """
    if isinstance(v, StringDict):
        return v
    if isinstance(v, dict):
        return StringDict(v)
    raise TypeError("Expected a StringDict or a dict for `values`")

RestClientConfig ¶

Bases: BaseModel

Rest client for given base URL. Includes transport and authentication configuration

base_url ¶

base_url: str

Base URL, typically server or API root. All endpoints with the same base URL share the same authentication.

Example: https://api.example.com/v1

timeout ¶

timeout: float | None = None

Request timeout in seconds. Default is 5. Set 0 to disable timout.

RestErrorHandler ¶

Bases: BaseModel

condition ¶

condition: str | None = None

JMESPath expression to look for in the response body. Error will be generated if expression evaluates to True

error_status_codes ¶

error_status_codes: list[int] = []

List of HTTP status codes to be treated as errors.

message ¶

message: str | None = None

JMESPath expression to extract error message from respose. If omitted entire response will be included in error.

RestResponse ¶

Bases: BaseModel

Response configuration. Response can be paged, polled URL or in body.

content_type ¶

content_type: DataType

Returned data type

locator ¶

locator: str | None = None

JMESPath to read data from JSON body. If not set then entire body is treated as data.

StatePoller ¶

Bases: BaseModel

A response handler for state-based polling APIs.

This handler is designed for asynchronous workflows where the client repeatedly polls an endpoint until a certain state is reached (e.g., job completion, resource readiness). Once the condition is met, the pipeline continues by reading from the final data locator.

kind ¶

kind: Literal[STATEPOLLING]

Specifies the handler type as StatePolling.

ready_status ¶

ready_status: str

JMESPath to read status from response. If the value at path evaluates to True then no more requests are made, and the API tries to read data from URL specified by locator.

URLPoller ¶

Bases: BaseModel

URL Poller makes request(s) to remote API until an URL is returned

kind ¶

kind: Literal[URLPOLLING]

Specifies the handler type as URLPolling.

ready_status ¶

ready_status: str | None = None

JMESPath to read status from response. If the value at path evaluates to True then no more requests are made, and the API tries to read data from URL specified by locator.

Pipeline API

__all__ ¶

BigQueryConfig ¶

credential_file ¶

dataset ¶

project ¶

region ¶

BigQueryConnection ¶

data_mode ¶

key ¶

kind ¶

partition ¶

schema_mode ¶

version ¶

BucketConfig ¶

bucket ¶

data_prefix ¶

locator_wildcard ¶

region ¶

CSVConnection ¶

ClickhouseConfig ¶

blocksize ¶

ClickhouseConnection ¶

key ¶

kind ¶

version ¶

Column ¶

name ¶

type ¶

Columns ¶

Connection ¶

config ¶

create_statement ¶

fields ¶

kind ¶

params ¶

show_schema ¶

ConnectionConfiguration ¶

bigquery ¶

clickhouse ¶

gs ¶

llm ¶

local ¶

s3 ¶

CustomConnection ¶

classname ¶

module ¶

params ¶

load ¶

DeltatableConnection ¶

data_mode ¶

kind ¶

optimize ¶

Deltatable¶

partition ¶

schema_mode ¶

writer_features ¶

Dimension ¶

load ¶

EphemeralConnection ¶

FileConnection ¶

GSConfig ¶

credential_file ¶

hmac_key ¶

hmac_secret ¶

JSONConnection ¶

OAuth2Provider ¶

access_token ¶

name ¶

on_refresh_fail ¶

on_token_refresh ¶

refresh_token ¶

subject_token ¶

ParquetConnection ¶

PhysicalConnection ¶

locator ¶

raw_dispatch ¶

RestConnection ¶

client ¶

fields ¶

all ¶