flanks-client 0.1.0

Python SDK for Flanks API

Flanks Python SDK

Async Python client for the Flanks API.

Features

• Fully typed - Complete type hints with strict mypy validation
• Async first - Built on httpx with async/await throughout
• Automatic authentication - OAuth2 token lifecycle handled transparently
• Smart retries - Exponential backoff for server errors, automatic token refresh on 401
• Pagination helpers - Async iterators for paginated endpoints
• Comprehensive error handling - Specific exceptions for different error types

Installation

pip install flanks-client

Quick Start

import asyncio
from flanks import FlanksClient

async def main():
    # Uses FLANKS_CLIENT_ID and FLANKS_CLIENT_SECRET env vars
    async with FlanksClient() as client:
        # List banking entities
        entities = await client.entities.list()
        for entity in entities:
            print(f"{entity.name} ({entity.country})")

        # Iterate through sessions with automatic pagination
        async for session in client.connect.list_sessions():
            print(f"{session.session_id}: {session.status}")

asyncio.run(main())

Configuration

from flanks import FlanksClient

client = FlanksClient(
    client_id="your_client_id",       # or FLANKS_CLIENT_ID env var
    client_secret="your_secret",       # or FLANKS_CLIENT_SECRET env var
    base_url="https://api.flanks.io",  # default
    timeout=60.0,                      # request timeout in seconds
    retries=1,                         # retry count for 5xx errors
    retry_backoff=1.0,                 # exponential backoff base (seconds)
    version="2026-01-01",              # API version date
)

API Clients

Entities

# List all available banking entities
entities = await client.entities.list()

Connect (v2)

from flanks.connect import SessionQuery, SessionStatus, SessionConfig

# Iterate all sessions
async for session in client.connect.list_sessions():
    print(session)

# Filter sessions by status
query = SessionQuery(status_in=[SessionStatus.FINISHED_OK])
async for session in client.connect.list_sessions(query):
    print(session)

# Manual pagination
page = await client.connect.list_sessions_page()
print(page.items)
if page.has_next():
    next_page = await client.connect.list_sessions_page(page_token=page.next_page_token)

# Create session
config = SessionConfig(connector_id="connector_123")
session = await client.connect.create_session(config)

# List connectors
async for connector in client.connect.list_connectors():
    print(connector.name)

Credentials

# Get credential status
status = await client.credentials.get_status("creds_token")

# List all credentials (page-based)
page1 = await client.credentials.list(page=1)

# Force SCA
await client.credentials.force_sca("creds_token")

# Delete credentials
await client.credentials.delete("creds_token")

Aggregation (v1)

# Get portfolios
portfolios = await client.aggregation_v1.get_portfolios(
    credentials_token="creds_token",
    date_from="2024-01-01"
)

# Get accounts
accounts = await client.aggregation_v1.get_accounts("creds_token")

# Get transactions
transactions = await client.aggregation_v1.get_account_transactions(
    credentials_token="creds_token",
    date_from="2024-01-01",
    date_to="2024-12-31"
)

# Get investments
investments = await client.aggregation_v1.get_investments("creds_token")
investment_txns = await client.aggregation_v1.get_investment_transactions("creds_token")

# Get liabilities (loans, mortgages)
liabilities = await client.aggregation_v1.get_liabilities("creds_token")
liability_txns = await client.aggregation_v1.get_liability_transactions("creds_token")

# Get cards
cards = await client.aggregation_v1.get_cards("creds_token")
card_txns = await client.aggregation_v1.get_card_transactions("creds_token")

# Get identity
identity = await client.aggregation_v1.get_identity("creds_token")

# Get holders
holders = await client.aggregation_v1.get_holders("creds_token")

Error Handling

from flanks import (
    FlanksAuthError,
    FlanksValidationError,
    FlanksNotFoundError,
    FlanksServerError,
    FlanksNetworkError,
)

try:
    entities = await client.entities.list()
except FlanksAuthError as e:
    # 401/403 - Invalid credentials or expired token
    print(f"Auth failed: {e}")
except FlanksValidationError as e:
    # 400 - Invalid request parameters
    print(f"Validation error: {e.response_body}")
except FlanksNotFoundError as e:
    # 404 - Resource not found
    print(f"Not found: {e}")
except FlanksServerError as e:
    # 5xx - Server error (after retries exhausted)
    print(f"Server error: {e.status_code}")
except FlanksNetworkError as e:
    # Network-level failure (connection refused, timeout, DNS)
    print(f"Network error: {e}")

Advanced Usage

Context Manager

# Automatic resource cleanup
async with FlanksClient() as client:
    entities = await client.entities.list()
# HTTP client closed automatically

Manual Resource Management

client = FlanksClient()
try:
    entities = await client.entities.list()
finally:
    await client.close()

Direct Transport Access

# For endpoints not yet wrapped by the SDK
response = await client.transport.api_call( "/custom/endpoint", {"param": "value"} )

Development

Running Tests

# Run all tests
uv run pytest

# Run with verbose output
uv run pytest -v

# Run specific test file
uv run pytest tests/test_client.py

Type Checking

uv run ty check

Linting

uv run ruff check flanks tests
uv run ruff format flanks tests

Requirements

• Python 3.10+
• httpx 0.28.1+
• pydantic 2.10+

License

MIT