Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,10 @@

- Added `form4api.__version__` (read from the installed package metadata, so it
always equals the pyproject version).
- Responses with fields added by newer API versions no longer raise
`TypeError` in companies/insiders/signals/webhooks methods (sync and async).
Unknown keys are now ignored, as the transactions, search and generated
models already did.

## 0.9.0 — 2026-10-05

Expand Down
27 changes: 21 additions & 6 deletions form4api/_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,21 @@
from dataclasses import dataclass, fields


class _FromDictMixin:
"""Tolerant construction for flat dataclasses built from API payloads.

The API only ever adds fields, and a dataclass constructor rejects unknown
keyword arguments, so `Model(**data)` crashes on the first new field.
`_from_dict` drops keys the dataclass does not declare. Key conversion
(camelCase to snake_case) already happened in the client before this runs.
Models with nested dataclasses override `_from_dict` to hydrate them."""

@classmethod
def _from_dict(cls, data: dict):
known = {f.name for f in fields(cls)} # type: ignore[arg-type]
return cls(**{k: v for k, v in data.items() if k in known})


@dataclass
class TopHolder:
manager_cik: str | None = None
Expand Down Expand Up @@ -88,7 +103,7 @@ def _from_dict(cls, data: dict) -> "Transaction":


@dataclass
class Insider:
class Insider(_FromDictMixin):
cik: str
name: str
is_director: bool
Expand All @@ -99,7 +114,7 @@ class Insider:


@dataclass
class Company:
class Company(_FromDictMixin):
cik: str
name: str
ticker: str | None
Expand Down Expand Up @@ -155,7 +170,7 @@ def _from_dict(cls, data: dict) -> "SearchResults":


@dataclass
class InsiderSignal:
class InsiderSignal(_FromDictMixin):
ticker: str | None
company_name: str
signal_date: str
Expand All @@ -166,7 +181,7 @@ class InsiderSignal:


@dataclass
class WebhookCreated:
class WebhookCreated(_FromDictMixin):
subscription_id: int
url: str
event_types: list[str]
Expand All @@ -176,7 +191,7 @@ class WebhookCreated:


@dataclass
class WebhookSubscription:
class WebhookSubscription(_FromDictMixin):
subscription_id: int
url: str
event_types: list[str]
Expand All @@ -185,7 +200,7 @@ class WebhookSubscription:


@dataclass
class WebhookEvent:
class WebhookEvent(_FromDictMixin):
delivery_id: int
subscription_id: int
event_type: str
Expand Down
8 changes: 4 additions & 4 deletions form4api/resources/_companies.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,11 @@ def __init__(self, client: Form4ApiClient) -> None:

def get(self, ticker: str) -> Company:
data = self._client._get(f"/v1/companies/{ticker}")
return Company(**data)
return Company._from_dict(data)

def insiders(self, ticker: str) -> list[Insider]:
data = self._client._get(f"/v1/companies/{ticker}/insiders")
return [Insider(**item) for item in data]
return [Insider._from_dict(item) for item in data]


class AsyncCompaniesResource(GeneratedAsyncCompaniesResource):
Expand All @@ -41,11 +41,11 @@ def __init__(self, client: AsyncForm4ApiClient) -> None:

async def get(self, ticker: str) -> Company:
data = await self._client._get(f"/v1/companies/{ticker}")
return Company(**data)
return Company._from_dict(data)

async def insiders(self, ticker: str) -> list[Insider]:
data = await self._client._get(f"/v1/companies/{ticker}/insiders")
return [Insider(**item) for item in data]
return [Insider._from_dict(item) for item in data]



8 changes: 4 additions & 4 deletions form4api/resources/_insiders.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,11 +39,11 @@ def search(
per_page: int = 20,
) -> list[Insider]:
data = self._client._get("/v1/insiders", _search_params(name, page, per_page))
return [Insider(**item) for item in data]
return [Insider._from_dict(item) for item in data]

def get(self, cik: str) -> Insider:
data = self._client._get(f"/v1/insiders/{cik}")
return Insider(**data)
return Insider._from_dict(data)

def transactions(
self,
Expand All @@ -67,11 +67,11 @@ def __init__(self, client: AsyncForm4ApiClient) -> None:

async def search(self, name: str, *, page: int = 1, per_page: int = 20) -> list[Insider]:
data = await self._client._get("/v1/insiders", _search_params(name, page, per_page))
return [Insider(**item) for item in data]
return [Insider._from_dict(item) for item in data]

async def get(self, cik: str) -> Insider:
data = await self._client._get(f"/v1/insiders/{cik}")
return Insider(**data)
return Insider._from_dict(data)

async def transactions(
self,
Expand Down
4 changes: 2 additions & 2 deletions form4api/resources/_signals.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ def list(
) -> list[InsiderSignal]:
params = _list_params(ticker, cluster_buy, cluster_sell, page, per_page)
data = self._client._get("/v1/signals", params)
return [InsiderSignal(**item) for item in data]
return [InsiderSignal._from_dict(item) for item in data]

def paginate(
self,
Expand Down Expand Up @@ -117,7 +117,7 @@ async def list(
) -> list[InsiderSignal]:
params = _list_params(ticker, cluster_buy, cluster_sell, page, per_page)
data = await self._client._get("/v1/signals", params)
return [InsiderSignal(**item) for item in data]
return [InsiderSignal._from_dict(item) for item in data]

async def paginate(
self,
Expand Down
12 changes: 6 additions & 6 deletions form4api/resources/_webhooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,11 @@ def __init__(self, client: Form4ApiClient) -> None:

def create(self, url: str, event_types: list[str]) -> WebhookCreated:
data = self._client._post("/v1/webhooks", {"url": url, "eventTypes": event_types})
return WebhookCreated(**data)
return WebhookCreated._from_dict(data)

def list(self) -> list[WebhookSubscription]:
data = self._client._get("/v1/webhooks")
return [WebhookSubscription(**item) for item in data]
return [WebhookSubscription._from_dict(item) for item in data]

def delete(self, subscription_id: int) -> None:
self._client._delete(f"/v1/webhooks/{subscription_id}")
Expand All @@ -28,7 +28,7 @@ def events(self, *, since: str | None = None) -> list[WebhookEvent]:
if since is not None:
params["since"] = since
data = self._client._get("/v1/webhooks/events", params or None)
return [WebhookEvent(**item) for item in data]
return [WebhookEvent._from_dict(item) for item in data]


class AsyncWebhooksResource:
Expand All @@ -39,11 +39,11 @@ def __init__(self, client: AsyncForm4ApiClient) -> None:

async def create(self, url: str, event_types: list[str]) -> WebhookCreated:
data = await self._client._post("/v1/webhooks", {"url": url, "eventTypes": event_types})
return WebhookCreated(**data)
return WebhookCreated._from_dict(data)

async def list(self) -> list[WebhookSubscription]:
data = await self._client._get("/v1/webhooks")
return [WebhookSubscription(**item) for item in data]
return [WebhookSubscription._from_dict(item) for item in data]

async def delete(self, subscription_id: int) -> None:
await self._client._delete(f"/v1/webhooks/{subscription_id}")
Expand All @@ -53,4 +53,4 @@ async def events(self, *, since: str | None = None) -> list[WebhookEvent]:
if since is not None:
params["since"] = since
data = await self._client._get("/v1/webhooks/events", params or None)
return [WebhookEvent(**item) for item in data]
return [WebhookEvent._from_dict(item) for item in data]
Loading
Loading