Client and adapters

The collect_* functions are enough for serial workloads. For anything that needs concurrency, rate limiting, or direct access to the REST endpoints, use the types on this page.

AsyncCompatibleClient

from ffiec_data_connect import AsyncCompatibleClient, OAuth2Credentials

AsyncCompatibleClient(
    credentials: OAuth2Credentials,
    max_concurrent: int = 5,
    rate_limit: float | None = 10,   # requests per second
    executor: ThreadPoolExecutor | None = None,
)

A thin wrapper over the collect_* functions that adds:

• a shared RateLimiter (disable with rate_limit=None),
• a ThreadPoolExecutor for parallel sync calls,
• async/await methods that offload to that executor,
• sync and async context-manager protocols (with / async with).

The client infers REST vs SOAP from the credential type. Since WebserviceCredentials now raises on construction, in practice this is always the REST path — but the conditional is still in the code, so if you ever see branches referencing _is_rest_client in a stack trace, that’s why.

Sync methods

Method	Purpose
collect_data(period, rssd_id, …)	Single call, rate-limited. Signature matches the module function minus creds.
collect_reporting_periods(…)	Single call, rate-limited.
collect_data_parallel(period, rssd_ids, …, progress_callback=None)	Many RSSDs at one period, in parallel. Returns {rssd_id: result}.
collect_time_series(rssd_id, periods, …)	One RSSD across many periods, in parallel.

Errors per item are returned as {"error": str(e), "rssd_id": ...} in the result dict rather than raising — this is deliberate so a single bad RSSD doesn’t abort a 5,000-bank batch. Inspect the dict for the "error" key.

from ffiec_data_connect import AsyncCompatibleClient, OAuth2Credentials

creds = OAuth2Credentials(username="...", bearer_token="eyJ...")

with AsyncCompatibleClient(creds, max_concurrent=8, rate_limit=5) as client:
    results = client.collect_data_parallel(
        reporting_period="12/31/2025",
        rssd_ids=["480228", "37", "852320"],
        series="call",
        output_type="pandas",
    )

for rssd_id, df in results.items():
    if isinstance(df, dict) and "error" in df:
        print(f"{rssd_id}: FAILED — {df['error']}")
    else:
        print(f"{rssd_id}: {len(df)} rows")

Async methods

Method	Purpose
await collect_data_async(...)	Single call, awaitable.
await collect_batch_async(period, rssd_ids, …, progress_callback=None)	Async fan-out with asyncio.Semaphore(max_concurrent).
await collect_time_series_async(rssd_id, periods, …)	Async time-series fan-out.

Internally these call loop.run_in_executor(self.executor, ...) — the underlying HTTP is still sync via httpx.Client, but the coroutine scheduling frees your event loop to do other work. If you’re already inside an asyncio app (FastAPI, aiohttp), prefer these over the sync variants.

async with AsyncCompatibleClient(creds, max_concurrent=10, rate_limit=5) as client:
    results = await client.collect_batch_async(
        reporting_period="12/31/2025",
        rssd_ids=rssd_ids,
        series="call",
        output_type="pandas",
    )

progress_callback accepts either a regular function or a coroutine function — the client uses asyncio.iscoroutinefunction() to decide whether to await it.

Resource cleanup

close() shuts down the internal executor if the client created it (it leaves externally-supplied executors alone). Both __exit__ and __aexit__ call it. Always use a context manager or call close() explicitly — dangling thread pools will prevent the interpreter from exiting cleanly.

See Async and rate limits for patterns and rate-limit tuning.

RateLimiter

Two distinct RateLimiter classes exist in the library, and both are exported. They serve different purposes.

The simple one (async_compatible.RateLimiter)

from ffiec_data_connect import RateLimiter  # this one is re-exported

RateLimiter(calls_per_second: float = 10)

Thread-safe per-second smoothing only. wait_if_needed() sleeps; the async variant await async_wait_if_needed() uses asyncio.sleep. This is what AsyncCompatibleClient constructs internally.

limiter = RateLimiter(calls_per_second=5)
for rssd_id in rssd_ids:
    limiter.wait_if_needed()
    do_work(rssd_id)

The hourly one (protocol_adapter.RateLimiter)

ffiec_data_connect.protocol_adapter.RateLimiter enforces both a per-second minimum interval AND a 2500-calls-per-hour quota (FFIEC’s published limit). It’s the one installed inside RESTAdapter. If you hit the hourly cap, wait_if_needed() blocks until the oldest call rolls off the hour window.

from ffiec_data_connect.protocol_adapter import RateLimiter as ApiRateLimiter

limiter = ApiRateLimiter(calls_per_hour=2500, calls_per_second=0.69)
limiter.get_stats()
# {"calls_this_hour": 0, "hourly_limit": 2500, "hourly_remaining": 2500,
#  "per_second_limit": 0.69, "last_call_seconds_ago": None}

Unless you’re building your own adapter on top of protocol_adapter, you want the top-level RateLimiter, not this one.

ProtocolAdapter

Abstract base class defining the seven operations every adapter must implement:

class ProtocolAdapter(ABC):
    def retrieve_reporting_periods(self, series: str) -> list[str]: ...
    def retrieve_facsimile(self, rssd_id, reporting_period, series) -> bytes: ...
    def retrieve_panel_of_reporters(self, reporting_period) -> list[dict]: ...
    def retrieve_filers_since_date(self, reporting_period, since_date) -> list[str]: ...
    def retrieve_filers_submission_datetime(self, reporting_period, since_date=None) -> list[dict]: ...
    def retrieve_ubpr_reporting_periods(self) -> list[str]: ...
    def retrieve_ubpr_xbrl_facsimile(self, rssd_id, reporting_period) -> bytes: ...

    @property
    def protocol_name(self) -> str: ...

Subclass it only if you’re writing a test double or an alternative transport. The production implementation is RESTAdapter.

RESTAdapter

from ffiec_data_connect import RESTAdapter, OAuth2Credentials

RESTAdapter(
    credentials: OAuth2Credentials,
    rate_limiter: RateLimiter | None = None,   # from protocol_adapter
    session: httpx.Client | None = None,
)

The real work. Talks to https://ffieccdr.azure-api.us/public with httpx, handles retries, applies the hourly rate limiter, validates response payloads via Pydantic models, then hands the data to DataNormalizer.

You rarely construct this directly — go through create_protocol_adapter or just use the collect_* functions. Instantiate it yourself when:

• You need a custom httpx.Client (mTLS, corporate proxy, custom timeouts).
• You want to call one of the retrieve_* methods directly because the collect_* surface doesn’t expose the shape you need.

import httpx
from ffiec_data_connect import OAuth2Credentials, RESTAdapter

client = httpx.Client(timeout=60.0, verify="/etc/ssl/corp-ca.pem")
adapter = RESTAdapter(
    OAuth2Credentials(username="...", bearer_token="eyJ..."),
    session=client,
)
periods = adapter.retrieve_reporting_periods("call")

RESTAdapter.BASE_URL and RESTAdapter.DEFAULT_RATE_LIMIT are class-level constants exposed for override/inspection.

create_protocol_adapter

from ffiec_data_connect import create_protocol_adapter

adapter = create_protocol_adapter(
    credentials: OAuth2Credentials,
    session: httpx.Client | None = None,  # optional
)

Factory. Given OAuth2Credentials, returns a configured RESTAdapter. If you pass WebserviceCredentials, it raises SOAPDeprecationError. If you pass anything else, it raises CredentialError.

Prefer this over constructing RESTAdapter directly — it’s where future adapter dispatch (e.g., a mock adapter for testing) would plug in.

SOAPAdapter (deprecated stub)

Exported for symmetry with v2. Its __init__ raises SOAPDeprecationError unconditionally. Do not use.

See also

• Async and rate limits — practical concurrency patterns.
• Functions — the high-level façade over RESTAdapter.
• Upstream async_compatible.py and protocol_adapter.py.