From f9120310bf8e72e469d74d9da5a47ab3b5873fa4 Mon Sep 17 00:00:00 2001 From: "Joseph T. French" Date: Tue, 19 May 2026 20:45:53 -0500 Subject: [PATCH] Add models for execute event block response and rollforward mechanics - Introduced `ExecuteEventBlockResponse` and `ExecuteEventBlockResponseQbErrorType0` models to handle the response structure of the execute event block API. - Added `JournalEntryLineItemInputMetadataType0` and `LineItemMetadataPredicate` models for enhanced metadata handling in journal entries. - Created `OperationEnvelopeExecuteEventBlockResponse` and `OperationEnvelopeExecuteEventBlockResponseStatus` to manage operation envelope responses. - Implemented `RollforwardMechanics` and `RollforwardMechanicsValidationMode` for rollforward attribution mechanics. - Added `UpdateRollforwardArm` and `UpdateRollforwardRequest` models to facilitate updates to rollforward blocks, including validation modes. - Introduced `UpdateRollforwardRequestValidationModeType0` for validation mode handling in rollforward requests. --- .../op_create_information_block.py | 29 +- .../op_delete_information_block.py | 29 +- .../op_execute_event_block.py | 314 ++++++++++++++++++ .../op_update_information_block.py | 29 +- robosystems_client/models/__init__.py | 48 +++ .../models/artifact_response.py | 27 +- .../models/attribution_filter.py | 121 +++++++ .../models/create_rollforward_arm.py | 88 +++++ .../models/create_rollforward_request.py | 176 ++++++++++ ...ate_rollforward_request_validation_mode.py | 10 + .../models/delete_rollforward_arm.py | 85 +++++ .../models/delete_rollforward_request.py | 66 ++++ .../models/execute_event_block_request.py | 97 ++++++ .../models/execute_event_block_response.py | 144 ++++++++ ...te_event_block_response_qb_error_type_0.py | 47 +++ .../models/journal_entry_line_item_input.py | 52 ++- ...l_entry_line_item_input_metadata_type_0.py | 47 +++ .../models/line_item_metadata_predicate.py | 98 ++++++ ...n_envelope_execute_event_block_response.py | 158 +++++++++ ...ope_execute_event_block_response_status.py | 10 + .../models/rollforward_mechanics.py | 201 +++++++++++ .../rollforward_mechanics_validation_mode.py | 10 + .../models/update_rollforward_arm.py | 95 ++++++ .../models/update_rollforward_request.py | 208 ++++++++++++ ...lforward_request_validation_mode_type_0.py | 10 + 25 files changed, 2154 insertions(+), 45 deletions(-) create mode 100644 robosystems_client/api/extensions_robo_ledger/op_execute_event_block.py create mode 100644 robosystems_client/models/attribution_filter.py create mode 100644 robosystems_client/models/create_rollforward_arm.py create mode 100644 robosystems_client/models/create_rollforward_request.py create mode 100644 robosystems_client/models/create_rollforward_request_validation_mode.py create mode 100644 robosystems_client/models/delete_rollforward_arm.py create mode 100644 robosystems_client/models/delete_rollforward_request.py create mode 100644 robosystems_client/models/execute_event_block_request.py create mode 100644 robosystems_client/models/execute_event_block_response.py create mode 100644 robosystems_client/models/execute_event_block_response_qb_error_type_0.py create mode 100644 robosystems_client/models/journal_entry_line_item_input_metadata_type_0.py create mode 100644 robosystems_client/models/line_item_metadata_predicate.py create mode 100644 robosystems_client/models/operation_envelope_execute_event_block_response.py create mode 100644 robosystems_client/models/operation_envelope_execute_event_block_response_status.py create mode 100644 robosystems_client/models/rollforward_mechanics.py create mode 100644 robosystems_client/models/rollforward_mechanics_validation_mode.py create mode 100644 robosystems_client/models/update_rollforward_arm.py create mode 100644 robosystems_client/models/update_rollforward_request.py create mode 100644 robosystems_client/models/update_rollforward_request_validation_mode_type_0.py diff --git a/robosystems_client/api/extensions_robo_ledger/op_create_information_block.py b/robosystems_client/api/extensions_robo_ledger/op_create_information_block.py index bb3e94c..1bbe18e 100644 --- a/robosystems_client/api/extensions_robo_ledger/op_create_information_block.py +++ b/robosystems_client/api/extensions_robo_ledger/op_create_information_block.py @@ -7,6 +7,7 @@ from ... import errors from ...client import AuthenticatedClient, Client from ...models.create_legacy_arm import CreateLegacyArm +from ...models.create_rollforward_arm import CreateRollforwardArm from ...models.create_schedule_arm import CreateScheduleArm from ...models.error_response import ErrorResponse from ...models.operation_envelope_information_block_envelope import ( @@ -18,7 +19,7 @@ def _get_kwargs( graph_id: str, *, - body: CreateLegacyArm | CreateScheduleArm, + body: CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> dict[str, Any]: headers: dict[str, Any] = {} @@ -34,6 +35,8 @@ def _get_kwargs( if isinstance(body, CreateScheduleArm): _kwargs["json"] = body.to_dict() + elif isinstance(body, CreateRollforwardArm): + _kwargs["json"] = body.to_dict() else: _kwargs["json"] = body.to_dict() @@ -112,7 +115,7 @@ def sync_detailed( graph_id: str, *, client: AuthenticatedClient, - body: CreateLegacyArm | CreateScheduleArm, + body: CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> Response[ErrorResponse | OperationEnvelopeInformationBlockEnvelope]: """Create Information Block @@ -127,8 +130,8 @@ def sync_detailed( Args: graph_id (str): idempotency_key (None | str | Unset): - body (CreateLegacyArm | CreateScheduleArm): Create an Information Block. The body is a - discriminated union on + body (CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm): Create an Information + Block. The body is a discriminated union on `block_type`: pick the arm matching the block type you want to create. The schedule arm carries a fully typed payload; statement and metric arms accept an untyped payload but currently return HTTP @@ -160,7 +163,7 @@ def sync( graph_id: str, *, client: AuthenticatedClient, - body: CreateLegacyArm | CreateScheduleArm, + body: CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> ErrorResponse | OperationEnvelopeInformationBlockEnvelope | None: """Create Information Block @@ -175,8 +178,8 @@ def sync( Args: graph_id (str): idempotency_key (None | str | Unset): - body (CreateLegacyArm | CreateScheduleArm): Create an Information Block. The body is a - discriminated union on + body (CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm): Create an Information + Block. The body is a discriminated union on `block_type`: pick the arm matching the block type you want to create. The schedule arm carries a fully typed payload; statement and metric arms accept an untyped payload but currently return HTTP @@ -203,7 +206,7 @@ async def asyncio_detailed( graph_id: str, *, client: AuthenticatedClient, - body: CreateLegacyArm | CreateScheduleArm, + body: CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> Response[ErrorResponse | OperationEnvelopeInformationBlockEnvelope]: """Create Information Block @@ -218,8 +221,8 @@ async def asyncio_detailed( Args: graph_id (str): idempotency_key (None | str | Unset): - body (CreateLegacyArm | CreateScheduleArm): Create an Information Block. The body is a - discriminated union on + body (CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm): Create an Information + Block. The body is a discriminated union on `block_type`: pick the arm matching the block type you want to create. The schedule arm carries a fully typed payload; statement and metric arms accept an untyped payload but currently return HTTP @@ -249,7 +252,7 @@ async def asyncio( graph_id: str, *, client: AuthenticatedClient, - body: CreateLegacyArm | CreateScheduleArm, + body: CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> ErrorResponse | OperationEnvelopeInformationBlockEnvelope | None: """Create Information Block @@ -264,8 +267,8 @@ async def asyncio( Args: graph_id (str): idempotency_key (None | str | Unset): - body (CreateLegacyArm | CreateScheduleArm): Create an Information Block. The body is a - discriminated union on + body (CreateLegacyArm | CreateRollforwardArm | CreateScheduleArm): Create an Information + Block. The body is a discriminated union on `block_type`: pick the arm matching the block type you want to create. The schedule arm carries a fully typed payload; statement and metric arms accept an untyped payload but currently return HTTP diff --git a/robosystems_client/api/extensions_robo_ledger/op_delete_information_block.py b/robosystems_client/api/extensions_robo_ledger/op_delete_information_block.py index 559f523..6bd261b 100644 --- a/robosystems_client/api/extensions_robo_ledger/op_delete_information_block.py +++ b/robosystems_client/api/extensions_robo_ledger/op_delete_information_block.py @@ -7,6 +7,7 @@ from ... import errors from ...client import AuthenticatedClient, Client from ...models.delete_legacy_arm import DeleteLegacyArm +from ...models.delete_rollforward_arm import DeleteRollforwardArm from ...models.delete_schedule_arm import DeleteScheduleArm from ...models.error_response import ErrorResponse from ...models.operation_envelope_delete_information_block_response import ( @@ -18,7 +19,7 @@ def _get_kwargs( graph_id: str, *, - body: DeleteLegacyArm | DeleteScheduleArm, + body: DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> dict[str, Any]: headers: dict[str, Any] = {} @@ -34,6 +35,8 @@ def _get_kwargs( if isinstance(body, DeleteScheduleArm): _kwargs["json"] = body.to_dict() + elif isinstance(body, DeleteRollforwardArm): + _kwargs["json"] = body.to_dict() else: _kwargs["json"] = body.to_dict() @@ -114,7 +117,7 @@ def sync_detailed( graph_id: str, *, client: AuthenticatedClient, - body: DeleteLegacyArm | DeleteScheduleArm, + body: DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> Response[ErrorResponse | OperationEnvelopeDeleteInformationBlockResponse]: """Delete Information Block @@ -129,8 +132,8 @@ def sync_detailed( Args: graph_id (str): idempotency_key (None | str | Unset): - body (DeleteLegacyArm | DeleteScheduleArm): Delete an Information Block. The body is a - discriminated union on + body (DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm): Delete an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed delete payload; statement and metric arms return HTTP 501. @@ -160,7 +163,7 @@ def sync( graph_id: str, *, client: AuthenticatedClient, - body: DeleteLegacyArm | DeleteScheduleArm, + body: DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> ErrorResponse | OperationEnvelopeDeleteInformationBlockResponse | None: """Delete Information Block @@ -175,8 +178,8 @@ def sync( Args: graph_id (str): idempotency_key (None | str | Unset): - body (DeleteLegacyArm | DeleteScheduleArm): Delete an Information Block. The body is a - discriminated union on + body (DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm): Delete an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed delete payload; statement and metric arms return HTTP 501. @@ -201,7 +204,7 @@ async def asyncio_detailed( graph_id: str, *, client: AuthenticatedClient, - body: DeleteLegacyArm | DeleteScheduleArm, + body: DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> Response[ErrorResponse | OperationEnvelopeDeleteInformationBlockResponse]: """Delete Information Block @@ -216,8 +219,8 @@ async def asyncio_detailed( Args: graph_id (str): idempotency_key (None | str | Unset): - body (DeleteLegacyArm | DeleteScheduleArm): Delete an Information Block. The body is a - discriminated union on + body (DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm): Delete an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed delete payload; statement and metric arms return HTTP 501. @@ -245,7 +248,7 @@ async def asyncio( graph_id: str, *, client: AuthenticatedClient, - body: DeleteLegacyArm | DeleteScheduleArm, + body: DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> ErrorResponse | OperationEnvelopeDeleteInformationBlockResponse | None: """Delete Information Block @@ -260,8 +263,8 @@ async def asyncio( Args: graph_id (str): idempotency_key (None | str | Unset): - body (DeleteLegacyArm | DeleteScheduleArm): Delete an Information Block. The body is a - discriminated union on + body (DeleteLegacyArm | DeleteRollforwardArm | DeleteScheduleArm): Delete an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed delete payload; statement and metric arms return HTTP 501. diff --git a/robosystems_client/api/extensions_robo_ledger/op_execute_event_block.py b/robosystems_client/api/extensions_robo_ledger/op_execute_event_block.py new file mode 100644 index 0000000..0a4200f --- /dev/null +++ b/robosystems_client/api/extensions_robo_ledger/op_execute_event_block.py @@ -0,0 +1,314 @@ +from http import HTTPStatus +from typing import Any +from urllib.parse import quote + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.error_response import ErrorResponse +from ...models.execute_event_block_request import ExecuteEventBlockRequest +from ...models.operation_envelope_execute_event_block_response import ( + OperationEnvelopeExecuteEventBlockResponse, +) +from ...types import UNSET, Response, Unset + + +def _get_kwargs( + graph_id: str, + *, + body: ExecuteEventBlockRequest, + idempotency_key: None | str | Unset = UNSET, +) -> dict[str, Any]: + headers: dict[str, Any] = {} + if not isinstance(idempotency_key, Unset): + headers["Idempotency-Key"] = idempotency_key + + _kwargs: dict[str, Any] = { + "method": "post", + "url": "/extensions/roboledger/{graph_id}/operations/execute-event-block".format( + graph_id=quote(str(graph_id), safe=""), + ), + } + + _kwargs["json"] = body.to_dict() + + headers["Content-Type"] = "application/json" + + _kwargs["headers"] = headers + return _kwargs + + +def _parse_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> ErrorResponse | OperationEnvelopeExecuteEventBlockResponse | None: + if response.status_code == 200: + response_200 = OperationEnvelopeExecuteEventBlockResponse.from_dict(response.json()) + + return response_200 + + if response.status_code == 400: + response_400 = ErrorResponse.from_dict(response.json()) + + return response_400 + + if response.status_code == 401: + response_401 = ErrorResponse.from_dict(response.json()) + + return response_401 + + if response.status_code == 403: + response_403 = ErrorResponse.from_dict(response.json()) + + return response_403 + + if response.status_code == 404: + response_404 = ErrorResponse.from_dict(response.json()) + + return response_404 + + if response.status_code == 409: + response_409 = ErrorResponse.from_dict(response.json()) + + return response_409 + + if response.status_code == 422: + response_422 = ErrorResponse.from_dict(response.json()) + + return response_422 + + if response.status_code == 429: + response_429 = ErrorResponse.from_dict(response.json()) + + return response_429 + + if response.status_code == 500: + response_500 = ErrorResponse.from_dict(response.json()) + + return response_500 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Response[ErrorResponse | OperationEnvelopeExecuteEventBlockResponse]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + graph_id: str, + *, + client: AuthenticatedClient, + body: ExecuteEventBlockRequest, + idempotency_key: None | str | Unset = UNSET, +) -> Response[ErrorResponse | OperationEnvelopeExecuteEventBlockResponse]: + """Execute Event Block (publish to source-of-truth system) + + For events on a connection with write_policy='qb_authoritative' or 'hybrid', publish the captured GL + plan to the source-of-truth system (QuickBooks). Captures qb_txn_id on + event.metadata.qb_external_id, transitions status to 'fulfilled' (or 'pending' on rejection), and + promotes draft GL rows to 'posted'. Native-policy events fast-path through with no QB write — + RoboSystems is the system of record. + + **Idempotency**: supply an `Idempotency-Key` header to make safe retries; replays within 24 hours + return the same envelope. Reusing the key with a different body returns HTTP 409 Conflict. + + Args: + graph_id (str): + idempotency_key (None | str | Unset): + body (ExecuteEventBlockRequest): Request to publish an event to the source-of-truth + system. + + For events on a connection with `write_policy='qb_authoritative'` + (or `'hybrid'`), this triggers a synchronous write to QuickBooks + via the QB API. The returned `qb_txn_id` lands on + `event.metadata.qb_external_id` and the event transitions to + `committed` (in flight) → `fulfilled` (QB accepted) or `pending` + (QB rejected). + + `'native'`-policy events fast-path through with no QB write — + RoboSystems IS the source of truth, no outbound publish needed. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[ErrorResponse | OperationEnvelopeExecuteEventBlockResponse] + """ + + kwargs = _get_kwargs( + graph_id=graph_id, + body=body, + idempotency_key=idempotency_key, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + graph_id: str, + *, + client: AuthenticatedClient, + body: ExecuteEventBlockRequest, + idempotency_key: None | str | Unset = UNSET, +) -> ErrorResponse | OperationEnvelopeExecuteEventBlockResponse | None: + """Execute Event Block (publish to source-of-truth system) + + For events on a connection with write_policy='qb_authoritative' or 'hybrid', publish the captured GL + plan to the source-of-truth system (QuickBooks). Captures qb_txn_id on + event.metadata.qb_external_id, transitions status to 'fulfilled' (or 'pending' on rejection), and + promotes draft GL rows to 'posted'. Native-policy events fast-path through with no QB write — + RoboSystems is the system of record. + + **Idempotency**: supply an `Idempotency-Key` header to make safe retries; replays within 24 hours + return the same envelope. Reusing the key with a different body returns HTTP 409 Conflict. + + Args: + graph_id (str): + idempotency_key (None | str | Unset): + body (ExecuteEventBlockRequest): Request to publish an event to the source-of-truth + system. + + For events on a connection with `write_policy='qb_authoritative'` + (or `'hybrid'`), this triggers a synchronous write to QuickBooks + via the QB API. The returned `qb_txn_id` lands on + `event.metadata.qb_external_id` and the event transitions to + `committed` (in flight) → `fulfilled` (QB accepted) or `pending` + (QB rejected). + + `'native'`-policy events fast-path through with no QB write — + RoboSystems IS the source of truth, no outbound publish needed. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + ErrorResponse | OperationEnvelopeExecuteEventBlockResponse + """ + + return sync_detailed( + graph_id=graph_id, + client=client, + body=body, + idempotency_key=idempotency_key, + ).parsed + + +async def asyncio_detailed( + graph_id: str, + *, + client: AuthenticatedClient, + body: ExecuteEventBlockRequest, + idempotency_key: None | str | Unset = UNSET, +) -> Response[ErrorResponse | OperationEnvelopeExecuteEventBlockResponse]: + """Execute Event Block (publish to source-of-truth system) + + For events on a connection with write_policy='qb_authoritative' or 'hybrid', publish the captured GL + plan to the source-of-truth system (QuickBooks). Captures qb_txn_id on + event.metadata.qb_external_id, transitions status to 'fulfilled' (or 'pending' on rejection), and + promotes draft GL rows to 'posted'. Native-policy events fast-path through with no QB write — + RoboSystems is the system of record. + + **Idempotency**: supply an `Idempotency-Key` header to make safe retries; replays within 24 hours + return the same envelope. Reusing the key with a different body returns HTTP 409 Conflict. + + Args: + graph_id (str): + idempotency_key (None | str | Unset): + body (ExecuteEventBlockRequest): Request to publish an event to the source-of-truth + system. + + For events on a connection with `write_policy='qb_authoritative'` + (or `'hybrid'`), this triggers a synchronous write to QuickBooks + via the QB API. The returned `qb_txn_id` lands on + `event.metadata.qb_external_id` and the event transitions to + `committed` (in flight) → `fulfilled` (QB accepted) or `pending` + (QB rejected). + + `'native'`-policy events fast-path through with no QB write — + RoboSystems IS the source of truth, no outbound publish needed. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[ErrorResponse | OperationEnvelopeExecuteEventBlockResponse] + """ + + kwargs = _get_kwargs( + graph_id=graph_id, + body=body, + idempotency_key=idempotency_key, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + graph_id: str, + *, + client: AuthenticatedClient, + body: ExecuteEventBlockRequest, + idempotency_key: None | str | Unset = UNSET, +) -> ErrorResponse | OperationEnvelopeExecuteEventBlockResponse | None: + """Execute Event Block (publish to source-of-truth system) + + For events on a connection with write_policy='qb_authoritative' or 'hybrid', publish the captured GL + plan to the source-of-truth system (QuickBooks). Captures qb_txn_id on + event.metadata.qb_external_id, transitions status to 'fulfilled' (or 'pending' on rejection), and + promotes draft GL rows to 'posted'. Native-policy events fast-path through with no QB write — + RoboSystems is the system of record. + + **Idempotency**: supply an `Idempotency-Key` header to make safe retries; replays within 24 hours + return the same envelope. Reusing the key with a different body returns HTTP 409 Conflict. + + Args: + graph_id (str): + idempotency_key (None | str | Unset): + body (ExecuteEventBlockRequest): Request to publish an event to the source-of-truth + system. + + For events on a connection with `write_policy='qb_authoritative'` + (or `'hybrid'`), this triggers a synchronous write to QuickBooks + via the QB API. The returned `qb_txn_id` lands on + `event.metadata.qb_external_id` and the event transitions to + `committed` (in flight) → `fulfilled` (QB accepted) or `pending` + (QB rejected). + + `'native'`-policy events fast-path through with no QB write — + RoboSystems IS the source of truth, no outbound publish needed. + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + ErrorResponse | OperationEnvelopeExecuteEventBlockResponse + """ + + return ( + await asyncio_detailed( + graph_id=graph_id, + client=client, + body=body, + idempotency_key=idempotency_key, + ) + ).parsed diff --git a/robosystems_client/api/extensions_robo_ledger/op_update_information_block.py b/robosystems_client/api/extensions_robo_ledger/op_update_information_block.py index 4cab27c..c2aa250 100644 --- a/robosystems_client/api/extensions_robo_ledger/op_update_information_block.py +++ b/robosystems_client/api/extensions_robo_ledger/op_update_information_block.py @@ -11,6 +11,7 @@ OperationEnvelopeInformationBlockEnvelope, ) from ...models.update_legacy_arm import UpdateLegacyArm +from ...models.update_rollforward_arm import UpdateRollforwardArm from ...models.update_schedule_arm import UpdateScheduleArm from ...types import UNSET, Response, Unset @@ -18,7 +19,7 @@ def _get_kwargs( graph_id: str, *, - body: UpdateLegacyArm | UpdateScheduleArm, + body: UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> dict[str, Any]: headers: dict[str, Any] = {} @@ -34,6 +35,8 @@ def _get_kwargs( if isinstance(body, UpdateScheduleArm): _kwargs["json"] = body.to_dict() + elif isinstance(body, UpdateRollforwardArm): + _kwargs["json"] = body.to_dict() else: _kwargs["json"] = body.to_dict() @@ -112,7 +115,7 @@ def sync_detailed( graph_id: str, *, client: AuthenticatedClient, - body: UpdateLegacyArm | UpdateScheduleArm, + body: UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> Response[ErrorResponse | OperationEnvelopeInformationBlockEnvelope]: """Update Information Block @@ -127,8 +130,8 @@ def sync_detailed( Args: graph_id (str): idempotency_key (None | str | Unset): - body (UpdateLegacyArm | UpdateScheduleArm): Update an Information Block. The body is a - discriminated union on + body (UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm): Update an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed update payload; statement and metric arms return HTTP 501 (statements are library-seeded; metric updates are @@ -159,7 +162,7 @@ def sync( graph_id: str, *, client: AuthenticatedClient, - body: UpdateLegacyArm | UpdateScheduleArm, + body: UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> ErrorResponse | OperationEnvelopeInformationBlockEnvelope | None: """Update Information Block @@ -174,8 +177,8 @@ def sync( Args: graph_id (str): idempotency_key (None | str | Unset): - body (UpdateLegacyArm | UpdateScheduleArm): Update an Information Block. The body is a - discriminated union on + body (UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm): Update an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed update payload; statement and metric arms return HTTP 501 (statements are library-seeded; metric updates are @@ -201,7 +204,7 @@ async def asyncio_detailed( graph_id: str, *, client: AuthenticatedClient, - body: UpdateLegacyArm | UpdateScheduleArm, + body: UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> Response[ErrorResponse | OperationEnvelopeInformationBlockEnvelope]: """Update Information Block @@ -216,8 +219,8 @@ async def asyncio_detailed( Args: graph_id (str): idempotency_key (None | str | Unset): - body (UpdateLegacyArm | UpdateScheduleArm): Update an Information Block. The body is a - discriminated union on + body (UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm): Update an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed update payload; statement and metric arms return HTTP 501 (statements are library-seeded; metric updates are @@ -246,7 +249,7 @@ async def asyncio( graph_id: str, *, client: AuthenticatedClient, - body: UpdateLegacyArm | UpdateScheduleArm, + body: UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm, idempotency_key: None | str | Unset = UNSET, ) -> ErrorResponse | OperationEnvelopeInformationBlockEnvelope | None: """Update Information Block @@ -261,8 +264,8 @@ async def asyncio( Args: graph_id (str): idempotency_key (None | str | Unset): - body (UpdateLegacyArm | UpdateScheduleArm): Update an Information Block. The body is a - discriminated union on + body (UpdateLegacyArm | UpdateRollforwardArm | UpdateScheduleArm): Update an Information + Block. The body is a discriminated union on `block_type` mirroring `CreateInformationBlockRequest`. The schedule arm carries a fully typed update payload; statement and metric arms return HTTP 501 (statements are library-seeded; metric updates are diff --git a/robosystems_client/models/__init__.py b/robosystems_client/models/__init__.py index 799619f..99e92f1 100644 --- a/robosystems_client/models/__init__.py +++ b/robosystems_client/models/__init__.py @@ -7,6 +7,7 @@ from .artifact_response import ArtifactResponse from .artifact_response_template_type_0 import ArtifactResponseTemplateType0 from .association_response import AssociationResponse +from .attribution_filter import AttributionFilter from .auth_response import AuthResponse from .auth_response_org_type_0 import AuthResponseOrgType0 from .auth_response_user import AuthResponseUser @@ -87,6 +88,11 @@ from .create_publish_list_request import CreatePublishListRequest from .create_report_request import CreateReportRequest from .create_repository_subscription_request import CreateRepositorySubscriptionRequest +from .create_rollforward_arm import CreateRollforwardArm +from .create_rollforward_request import CreateRollforwardRequest +from .create_rollforward_request_validation_mode import ( + CreateRollforwardRequestValidationMode, +) from .create_schedule_arm import CreateScheduleArm from .create_schedule_request import CreateScheduleRequest from .create_security_request import CreateSecurityRequest @@ -127,6 +133,8 @@ from .delete_publish_list_operation import DeletePublishListOperation from .delete_report_operation import DeleteReportOperation from .delete_result import DeleteResult +from .delete_rollforward_arm import DeleteRollforwardArm +from .delete_rollforward_request import DeleteRollforwardRequest from .delete_schedule_arm import DeleteScheduleArm from .delete_schedule_request import DeleteScheduleRequest from .delete_security_operation import DeleteSecurityOperation @@ -173,6 +181,11 @@ from .event_handler_response_transaction_template import ( EventHandlerResponseTransactionTemplate, ) +from .execute_event_block_request import ExecuteEventBlockRequest +from .execute_event_block_response import ExecuteEventBlockResponse +from .execute_event_block_response_qb_error_type_0 import ( + ExecuteEventBlockResponseQbErrorType0, +) from .fact_lite import FactLite from .fact_set_lite import FactSetLite from .file_info import FileInfo @@ -233,11 +246,15 @@ from .invoice_line_item import InvoiceLineItem from .invoices_response import InvoicesResponse from .journal_entry_line_item_input import JournalEntryLineItemInput +from .journal_entry_line_item_input_metadata_type_0 import ( + JournalEntryLineItemInputMetadataType0, +) from .journal_entry_line_item_response import JournalEntryLineItemResponse from .journal_entry_response import JournalEntryResponse from .ledger_agent_response import LedgerAgentResponse from .ledger_agent_response_address_type_0 import LedgerAgentResponseAddressType0 from .ledger_entity_response import LedgerEntityResponse +from .line_item_metadata_predicate import LineItemMetadataPredicate from .link_entity_taxonomy_request import LinkEntityTaxonomyRequest from .link_entity_taxonomy_request_basis import LinkEntityTaxonomyRequestBasis from .list_connections_provider_type_0 import ListConnectionsProviderType0 @@ -321,6 +338,12 @@ from .operation_envelope_event_handler_response_status import ( OperationEnvelopeEventHandlerResponseStatus, ) +from .operation_envelope_execute_event_block_response import ( + OperationEnvelopeExecuteEventBlockResponse, +) +from .operation_envelope_execute_event_block_response_status import ( + OperationEnvelopeExecuteEventBlockResponseStatus, +) from .operation_envelope_fiscal_calendar_response import ( OperationEnvelopeFiscalCalendarResponse, ) @@ -487,6 +510,8 @@ from .reset_password_validate_response import ResetPasswordValidateResponse from .response_mode import ResponseMode from .restore_backup_op import RestoreBackupOp +from .rollforward_mechanics import RollforwardMechanics +from .rollforward_mechanics_validation_mode import RollforwardMechanicsValidationMode from .rule_lite import RuleLite from .rule_target_lite import RuleTargetLite from .rule_variable_lite import RuleVariableLite @@ -627,6 +652,11 @@ from .update_password_request import UpdatePasswordRequest from .update_portfolio_block_operation import UpdatePortfolioBlockOperation from .update_publish_list_operation import UpdatePublishListOperation +from .update_rollforward_arm import UpdateRollforwardArm +from .update_rollforward_request import UpdateRollforwardRequest +from .update_rollforward_request_validation_mode_type_0 import ( + UpdateRollforwardRequestValidationModeType0, +) from .update_schedule_arm import UpdateScheduleArm from .update_schedule_request import UpdateScheduleRequest from .update_security_operation import UpdateSecurityOperation @@ -654,6 +684,7 @@ "ArtifactResponse", "ArtifactResponseTemplateType0", "AssociationResponse", + "AttributionFilter", "AuthResponse", "AuthResponseOrgType0", "AuthResponseUser", @@ -720,6 +751,9 @@ "CreatePublishListRequest", "CreateReportRequest", "CreateRepositorySubscriptionRequest", + "CreateRollforwardArm", + "CreateRollforwardRequest", + "CreateRollforwardRequestValidationMode", "CreateScheduleArm", "CreateScheduleRequest", "CreateSecurityRequest", @@ -756,6 +790,8 @@ "DeletePublishListOperation", "DeleteReportOperation", "DeleteResult", + "DeleteRollforwardArm", + "DeleteRollforwardRequest", "DeleteScheduleArm", "DeleteScheduleRequest", "DeleteSecurityOperation", @@ -794,6 +830,9 @@ "EventHandlerResponse", "EventHandlerResponseMatchMetadataExpressionType0", "EventHandlerResponseTransactionTemplate", + "ExecuteEventBlockRequest", + "ExecuteEventBlockResponse", + "ExecuteEventBlockResponseQbErrorType0", "FactLite", "FactSetLite", "FileInfo", @@ -844,11 +883,13 @@ "InvoiceLineItem", "InvoicesResponse", "JournalEntryLineItemInput", + "JournalEntryLineItemInputMetadataType0", "JournalEntryLineItemResponse", "JournalEntryResponse", "LedgerAgentResponse", "LedgerAgentResponseAddressType0", "LedgerEntityResponse", + "LineItemMetadataPredicate", "LinkEntityTaxonomyRequest", "LinkEntityTaxonomyRequestBasis", "ListConnectionsProviderType0", @@ -894,6 +935,8 @@ "OperationEnvelopeEventBlockEnvelopeStatus", "OperationEnvelopeEventHandlerResponse", "OperationEnvelopeEventHandlerResponseStatus", + "OperationEnvelopeExecuteEventBlockResponse", + "OperationEnvelopeExecuteEventBlockResponseStatus", "OperationEnvelopeFiscalCalendarResponse", "OperationEnvelopeFiscalCalendarResponseStatus", "OperationEnvelopeInformationBlockEnvelope", @@ -1000,6 +1043,8 @@ "ResetPasswordValidateResponse", "ResponseMode", "RestoreBackupOp", + "RollforwardMechanics", + "RollforwardMechanicsValidationMode", "RuleLite", "RuleTargetLite", "RuleVariableLite", @@ -1106,6 +1151,9 @@ "UpdatePasswordRequest", "UpdatePortfolioBlockOperation", "UpdatePublishListOperation", + "UpdateRollforwardArm", + "UpdateRollforwardRequest", + "UpdateRollforwardRequestValidationModeType0", "UpdateScheduleArm", "UpdateScheduleRequest", "UpdateSecurityOperation", diff --git a/robosystems_client/models/artifact_response.py b/robosystems_client/models/artifact_response.py index a322e5c..91f8513 100644 --- a/robosystems_client/models/artifact_response.py +++ b/robosystems_client/models/artifact_response.py @@ -11,6 +11,7 @@ if TYPE_CHECKING: from ..models.artifact_response_template_type_0 import ArtifactResponseTemplateType0 from ..models.metric_mechanics import MetricMechanics + from ..models.rollforward_mechanics import RollforwardMechanics from ..models.schedule_mechanics import ScheduleMechanics from ..models.statement_mechanics import StatementMechanics @@ -23,7 +24,7 @@ class ArtifactResponse: """The block's producible-artifact envelope — topic, template, mechanics. Attributes: - mechanics (MetricMechanics | ScheduleMechanics | StatementMechanics): + mechanics (MetricMechanics | RollforwardMechanics | ScheduleMechanics | StatementMechanics): topic (None | str | Unset): Structure.description — the block's human-readable topic. renderer_note (None | str | Unset): e.g. 'in thousands', 'except per share'. template (ArtifactResponseTemplateType0 | None | Unset): Reusable layout (ordering, subtotals, styling) when @@ -31,7 +32,9 @@ class ArtifactResponse: types. """ - mechanics: MetricMechanics | ScheduleMechanics | StatementMechanics + mechanics: ( + MetricMechanics | RollforwardMechanics | ScheduleMechanics | StatementMechanics + ) topic: None | str | Unset = UNSET renderer_note: None | str | Unset = UNSET template: ArtifactResponseTemplateType0 | None | Unset = UNSET @@ -39,6 +42,7 @@ class ArtifactResponse: def to_dict(self) -> dict[str, Any]: from ..models.artifact_response_template_type_0 import ArtifactResponseTemplateType0 + from ..models.metric_mechanics import MetricMechanics from ..models.schedule_mechanics import ScheduleMechanics from ..models.statement_mechanics import StatementMechanics @@ -47,6 +51,8 @@ def to_dict(self) -> dict[str, Any]: mechanics = self.mechanics.to_dict() elif isinstance(self.mechanics, StatementMechanics): mechanics = self.mechanics.to_dict() + elif isinstance(self.mechanics, MetricMechanics): + mechanics = self.mechanics.to_dict() else: mechanics = self.mechanics.to_dict() @@ -90,6 +96,7 @@ def to_dict(self) -> dict[str, Any]: def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: from ..models.artifact_response_template_type_0 import ArtifactResponseTemplateType0 from ..models.metric_mechanics import MetricMechanics + from ..models.rollforward_mechanics import RollforwardMechanics from ..models.schedule_mechanics import ScheduleMechanics from ..models.statement_mechanics import StatementMechanics @@ -97,7 +104,9 @@ def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: def _parse_mechanics( data: object, - ) -> MetricMechanics | ScheduleMechanics | StatementMechanics: + ) -> ( + MetricMechanics | RollforwardMechanics | ScheduleMechanics | StatementMechanics + ): try: if not isinstance(data, dict): raise TypeError() @@ -114,11 +123,19 @@ def _parse_mechanics( return mechanics_type_1 except (TypeError, ValueError, AttributeError, KeyError): pass + try: + if not isinstance(data, dict): + raise TypeError() + mechanics_type_2 = MetricMechanics.from_dict(data) + + return mechanics_type_2 + except (TypeError, ValueError, AttributeError, KeyError): + pass if not isinstance(data, dict): raise TypeError() - mechanics_type_2 = MetricMechanics.from_dict(data) + mechanics_type_3 = RollforwardMechanics.from_dict(data) - return mechanics_type_2 + return mechanics_type_3 mechanics = _parse_mechanics(d.pop("mechanics")) diff --git a/robosystems_client/models/attribution_filter.py b/robosystems_client/models/attribution_filter.py new file mode 100644 index 0000000..5fa1879 --- /dev/null +++ b/robosystems_client/models/attribution_filter.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.line_item_metadata_predicate import LineItemMetadataPredicate + + +T = TypeVar("T", bound="AttributionFilter") + + +@_attrs_define +class AttributionFilter: + """One flow-concept attribution rule on a rollforward IB. + + Pairs a target concept (the flow leaf the matched amount counts + toward) with a predicate (which LineItems match). The rollforward's + ``attribution_filters: list[AttributionFilter]`` declares every flow + the BS source decomposes into; the renderer evaluates them all per + period. + + ``target_element_id`` is resolved at create time from ``target_qname`` + via the rs-gaap library + tenant taxonomy lookup. Authors only need + to provide the qname; the element_id is filled in by the create + handler and the resolved value is what the envelope round-trips. + + Attributes: + target_qname (str): QName of the flow concept this filter produces facts for — e.g. ``rs- + gaap:ProceedsFromIssuanceOfCommonStock``. Resolved to ``target_element_id`` at create time. + predicate (LineItemMetadataPredicate): Filter ledger LineItems whose ``metadata_[field]`` is in ``values``. + + The single predicate kind shipped in Phase 2 MVP. Sufficient for any + source taxonomy that stamps a flow-tag column on each transaction + line — mini's ``TransactionDescriptionCode``, future XBRL GL + ``GenericFlowCategory`` columns, custom tenant tags. + + ``field`` is the JSONB key under ``line_items.metadata`` (e.g. + ``"transaction_description_code"``). ``values`` is the set of values + that route to the filter's target concept; matched LineItems aggregate + signed into the attributed fact for the period. + target_element_id (None | str | Unset): Resolved element id for ``target_qname``. Null at create time; populated + by the handler before persistence. Round-tripped in the envelope. + """ + + target_qname: str + predicate: LineItemMetadataPredicate + target_element_id: None | str | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + target_qname = self.target_qname + + predicate = self.predicate.to_dict() + + target_element_id: None | str | Unset + if isinstance(self.target_element_id, Unset): + target_element_id = UNSET + else: + target_element_id = self.target_element_id + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "target_qname": target_qname, + "predicate": predicate, + } + ) + if target_element_id is not UNSET: + field_dict["target_element_id"] = target_element_id + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.line_item_metadata_predicate import LineItemMetadataPredicate + + d = dict(src_dict) + target_qname = d.pop("target_qname") + + predicate = LineItemMetadataPredicate.from_dict(d.pop("predicate")) + + def _parse_target_element_id(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + target_element_id = _parse_target_element_id(d.pop("target_element_id", UNSET)) + + attribution_filter = cls( + target_qname=target_qname, + predicate=predicate, + target_element_id=target_element_id, + ) + + attribution_filter.additional_properties = d + return attribution_filter + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/create_rollforward_arm.py b/robosystems_client/models/create_rollforward_arm.py new file mode 100644 index 0000000..13b3054 --- /dev/null +++ b/robosystems_client/models/create_rollforward_arm.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +if TYPE_CHECKING: + from ..models.create_rollforward_request import CreateRollforwardRequest + + +T = TypeVar("T", bound="CreateRollforwardArm") + + +@_attrs_define +class CreateRollforwardArm: + """Create-information-block body for ``block_type="rollforward"``. + + Carries a typed rollforward payload. The block decomposes the period + change in a BS source element across the declared attribution + filters. + + Attributes: + block_type (Literal['rollforward']): Discriminator value selecting this arm. + payload (CreateRollforwardRequest): Create a rollforward Information Block. + + Mirrors :class:`CreateScheduleRequest` in shape. The block decomposes + the period change in ``bs_source_qname`` across the declared + attribution filters. Residual (Δ BS - Σ filter matches) falls back to + the default change tag — or, if no default is declared, surfaces as + an unattributed fact tagged with a synthetic residual concept. + """ + + block_type: Literal["rollforward"] + payload: CreateRollforwardRequest + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + block_type = self.block_type + + payload = self.payload.to_dict() + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "block_type": block_type, + "payload": payload, + } + ) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.create_rollforward_request import CreateRollforwardRequest + + d = dict(src_dict) + block_type = cast(Literal["rollforward"], d.pop("block_type")) + if block_type != "rollforward": + raise ValueError(f"block_type must match const 'rollforward', got '{block_type}'") + + payload = CreateRollforwardRequest.from_dict(d.pop("payload")) + + create_rollforward_arm = cls( + block_type=block_type, + payload=payload, + ) + + create_rollforward_arm.additional_properties = d + return create_rollforward_arm + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/create_rollforward_request.py b/robosystems_client/models/create_rollforward_request.py new file mode 100644 index 0000000..68dba7f --- /dev/null +++ b/robosystems_client/models/create_rollforward_request.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..models.create_rollforward_request_validation_mode import ( + CreateRollforwardRequestValidationMode, +) +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.attribution_filter import AttributionFilter + + +T = TypeVar("T", bound="CreateRollforwardRequest") + + +@_attrs_define +class CreateRollforwardRequest: + """Create a rollforward Information Block. + + Mirrors :class:`CreateScheduleRequest` in shape. The block decomposes + the period change in ``bs_source_qname`` across the declared + attribution filters. Residual (Δ BS - Σ filter matches) falls back to + the default change tag — or, if no default is declared, surfaces as + an unattributed fact tagged with a synthetic residual concept. + + Attributes: + name (str): Human-readable block name. + bs_source_qname (str): QName of the balance-sheet element whose period delta this block decomposes. Resolved to + ``bs_source_element_id`` at create time. + default_change_tag_qname (None | str | Unset): QName of the fallback flow concept (Tier 1 default change tag). + Residual amount — Δ BS minus the sum of filter-matched amounts — is attributed to this concept. When omitted, + residual surfaces unattributed; the validation_mode setting governs whether that's a hard error. + attribution_filters (list[AttributionFilter] | Unset): Filter predicates routing LineItems to flow concepts. + validation_mode (CreateRollforwardRequestValidationMode | Unset): How the renderer arbitrates when Σ filter + matches != Δ BS. ``strict`` raises; ``residual_as_default`` emits the residual as a default-tag fact (the common + case); ``warn_only`` logs and lets the imbalance pass. Default: + CreateRollforwardRequestValidationMode.RESIDUAL_AS_DEFAULT. + taxonomy_id (None | str | Unset): Owning taxonomy id (auto-resolved from ``bs_source_qname`` when omitted). + """ + + name: str + bs_source_qname: str + default_change_tag_qname: None | str | Unset = UNSET + attribution_filters: list[AttributionFilter] | Unset = UNSET + validation_mode: CreateRollforwardRequestValidationMode | Unset = ( + CreateRollforwardRequestValidationMode.RESIDUAL_AS_DEFAULT + ) + taxonomy_id: None | str | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + name = self.name + + bs_source_qname = self.bs_source_qname + + default_change_tag_qname: None | str | Unset + if isinstance(self.default_change_tag_qname, Unset): + default_change_tag_qname = UNSET + else: + default_change_tag_qname = self.default_change_tag_qname + + attribution_filters: list[dict[str, Any]] | Unset = UNSET + if not isinstance(self.attribution_filters, Unset): + attribution_filters = [] + for attribution_filters_item_data in self.attribution_filters: + attribution_filters_item = attribution_filters_item_data.to_dict() + attribution_filters.append(attribution_filters_item) + + validation_mode: str | Unset = UNSET + if not isinstance(self.validation_mode, Unset): + validation_mode = self.validation_mode.value + + taxonomy_id: None | str | Unset + if isinstance(self.taxonomy_id, Unset): + taxonomy_id = UNSET + else: + taxonomy_id = self.taxonomy_id + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "name": name, + "bs_source_qname": bs_source_qname, + } + ) + if default_change_tag_qname is not UNSET: + field_dict["default_change_tag_qname"] = default_change_tag_qname + if attribution_filters is not UNSET: + field_dict["attribution_filters"] = attribution_filters + if validation_mode is not UNSET: + field_dict["validation_mode"] = validation_mode + if taxonomy_id is not UNSET: + field_dict["taxonomy_id"] = taxonomy_id + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.attribution_filter import AttributionFilter + + d = dict(src_dict) + name = d.pop("name") + + bs_source_qname = d.pop("bs_source_qname") + + def _parse_default_change_tag_qname(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + default_change_tag_qname = _parse_default_change_tag_qname( + d.pop("default_change_tag_qname", UNSET) + ) + + _attribution_filters = d.pop("attribution_filters", UNSET) + attribution_filters: list[AttributionFilter] | Unset = UNSET + if _attribution_filters is not UNSET: + attribution_filters = [] + for attribution_filters_item_data in _attribution_filters: + attribution_filters_item = AttributionFilter.from_dict( + attribution_filters_item_data + ) + + attribution_filters.append(attribution_filters_item) + + _validation_mode = d.pop("validation_mode", UNSET) + validation_mode: CreateRollforwardRequestValidationMode | Unset + if isinstance(_validation_mode, Unset): + validation_mode = UNSET + else: + validation_mode = CreateRollforwardRequestValidationMode(_validation_mode) + + def _parse_taxonomy_id(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + taxonomy_id = _parse_taxonomy_id(d.pop("taxonomy_id", UNSET)) + + create_rollforward_request = cls( + name=name, + bs_source_qname=bs_source_qname, + default_change_tag_qname=default_change_tag_qname, + attribution_filters=attribution_filters, + validation_mode=validation_mode, + taxonomy_id=taxonomy_id, + ) + + create_rollforward_request.additional_properties = d + return create_rollforward_request + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/create_rollforward_request_validation_mode.py b/robosystems_client/models/create_rollforward_request_validation_mode.py new file mode 100644 index 0000000..60826f9 --- /dev/null +++ b/robosystems_client/models/create_rollforward_request_validation_mode.py @@ -0,0 +1,10 @@ +from enum import Enum + + +class CreateRollforwardRequestValidationMode(str, Enum): + RESIDUAL_AS_DEFAULT = "residual_as_default" + STRICT = "strict" + WARN_ONLY = "warn_only" + + def __str__(self) -> str: + return str(self.value) diff --git a/robosystems_client/models/delete_rollforward_arm.py b/robosystems_client/models/delete_rollforward_arm.py new file mode 100644 index 0000000..d20d138 --- /dev/null +++ b/robosystems_client/models/delete_rollforward_arm.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +if TYPE_CHECKING: + from ..models.delete_rollforward_request import DeleteRollforwardRequest + + +T = TypeVar("T", bound="DeleteRollforwardArm") + + +@_attrs_define +class DeleteRollforwardArm: + """Delete-information-block body for ``block_type="rollforward"``. + + Cascades through any synthetic facts produced by this block's filter + evaluations. The underlying ledger LineItems are not touched. + + Attributes: + block_type (Literal['rollforward']): Discriminator value selecting this arm. + payload (DeleteRollforwardRequest): Delete a rollforward block. + + Cascades through any synthetic facts produced by this block's filter + evaluations. The underlying ledger LineItems are not touched — only + the rollforward IB's projection of them. + """ + + block_type: Literal["rollforward"] + payload: DeleteRollforwardRequest + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + block_type = self.block_type + + payload = self.payload.to_dict() + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "block_type": block_type, + "payload": payload, + } + ) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.delete_rollforward_request import DeleteRollforwardRequest + + d = dict(src_dict) + block_type = cast(Literal["rollforward"], d.pop("block_type")) + if block_type != "rollforward": + raise ValueError(f"block_type must match const 'rollforward', got '{block_type}'") + + payload = DeleteRollforwardRequest.from_dict(d.pop("payload")) + + delete_rollforward_arm = cls( + block_type=block_type, + payload=payload, + ) + + delete_rollforward_arm.additional_properties = d + return delete_rollforward_arm + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/delete_rollforward_request.py b/robosystems_client/models/delete_rollforward_request.py new file mode 100644 index 0000000..3f57722 --- /dev/null +++ b/robosystems_client/models/delete_rollforward_request.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, TypeVar + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +T = TypeVar("T", bound="DeleteRollforwardRequest") + + +@_attrs_define +class DeleteRollforwardRequest: + """Delete a rollforward block. + + Cascades through any synthetic facts produced by this block's filter + evaluations. The underlying ledger LineItems are not touched — only + the rollforward IB's projection of them. + + Attributes: + structure_id (str): Structure ID of the rollforward block. + """ + + structure_id: str + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + structure_id = self.structure_id + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "structure_id": structure_id, + } + ) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + structure_id = d.pop("structure_id") + + delete_rollforward_request = cls( + structure_id=structure_id, + ) + + delete_rollforward_request.additional_properties = d + return delete_rollforward_request + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/execute_event_block_request.py b/robosystems_client/models/execute_event_block_request.py new file mode 100644 index 0000000..2f0e2ed --- /dev/null +++ b/robosystems_client/models/execute_event_block_request.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..types import UNSET, Unset + +T = TypeVar("T", bound="ExecuteEventBlockRequest") + + +@_attrs_define +class ExecuteEventBlockRequest: + """Request to publish an event to the source-of-truth system. + + For events on a connection with `write_policy='qb_authoritative'` + (or `'hybrid'`), this triggers a synchronous write to QuickBooks + via the QB API. The returned `qb_txn_id` lands on + `event.metadata.qb_external_id` and the event transitions to + `committed` (in flight) → `fulfilled` (QB accepted) or `pending` + (QB rejected). + + `'native'`-policy events fast-path through with no QB write — + RoboSystems IS the source of truth, no outbound publish needed. + + Attributes: + event_id (str): Event ID (`evt_*` ULID) to publish. The event's `metadata.connection_id` determines which QB + connection to write to; the connection's `write_policy` governs whether a write fires. + connection_id (None | str | Unset): Override for the connection to route the write through. Used by the close- + period batch path where schedule-originated events don't carry `connection_id` in their metadata. When unset, + the command reads `event.metadata.connection_id`. + """ + + event_id: str + connection_id: None | str | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + event_id = self.event_id + + connection_id: None | str | Unset + if isinstance(self.connection_id, Unset): + connection_id = UNSET + else: + connection_id = self.connection_id + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "event_id": event_id, + } + ) + if connection_id is not UNSET: + field_dict["connection_id"] = connection_id + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + event_id = d.pop("event_id") + + def _parse_connection_id(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + connection_id = _parse_connection_id(d.pop("connection_id", UNSET)) + + execute_event_block_request = cls( + event_id=event_id, + connection_id=connection_id, + ) + + execute_event_block_request.additional_properties = d + return execute_event_block_request + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/execute_event_block_response.py b/robosystems_client/models/execute_event_block_response.py new file mode 100644 index 0000000..b200c6f --- /dev/null +++ b/robosystems_client/models/execute_event_block_response.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.execute_event_block_response_qb_error_type_0 import ( + ExecuteEventBlockResponseQbErrorType0, + ) + + +T = TypeVar("T", bound="ExecuteEventBlockResponse") + + +@_attrs_define +class ExecuteEventBlockResponse: + """Outcome of an `execute-event-block` call. + + Attributes: + event_id (str): Echo of the event ID. + status (str): Post-execute event status. `'classified'` when no write fired (native policy or no-op). + `'committed'` when the QB write was in flight (intermediate state). `'fulfilled'` when QB accepted and local GL + drafts were promoted to posted. `'pending'` when QB rejected — see `qb_error` for the rejection detail; retry + after fixing the underlying issue. + qb_external_id (None | str | Unset): QB-side transaction ID returned by the JournalEntry API. Null when no write + fired (native policy) or when the write was rejected before getting an ID. + qb_error (ExecuteEventBlockResponseQbErrorType0 | None | Unset): QB rejection detail when status='pending'. + Shape: `{code, message, qb_response_at}`. Operator retries after fixing CoA mapping / amount validation / + closed-period. + """ + + event_id: str + status: str + qb_external_id: None | str | Unset = UNSET + qb_error: ExecuteEventBlockResponseQbErrorType0 | None | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + from ..models.execute_event_block_response_qb_error_type_0 import ( + ExecuteEventBlockResponseQbErrorType0, + ) + + event_id = self.event_id + + status = self.status + + qb_external_id: None | str | Unset + if isinstance(self.qb_external_id, Unset): + qb_external_id = UNSET + else: + qb_external_id = self.qb_external_id + + qb_error: dict[str, Any] | None | Unset + if isinstance(self.qb_error, Unset): + qb_error = UNSET + elif isinstance(self.qb_error, ExecuteEventBlockResponseQbErrorType0): + qb_error = self.qb_error.to_dict() + else: + qb_error = self.qb_error + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "event_id": event_id, + "status": status, + } + ) + if qb_external_id is not UNSET: + field_dict["qb_external_id"] = qb_external_id + if qb_error is not UNSET: + field_dict["qb_error"] = qb_error + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.execute_event_block_response_qb_error_type_0 import ( + ExecuteEventBlockResponseQbErrorType0, + ) + + d = dict(src_dict) + event_id = d.pop("event_id") + + status = d.pop("status") + + def _parse_qb_external_id(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + qb_external_id = _parse_qb_external_id(d.pop("qb_external_id", UNSET)) + + def _parse_qb_error( + data: object, + ) -> ExecuteEventBlockResponseQbErrorType0 | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, dict): + raise TypeError() + qb_error_type_0 = ExecuteEventBlockResponseQbErrorType0.from_dict(data) + + return qb_error_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(ExecuteEventBlockResponseQbErrorType0 | None | Unset, data) + + qb_error = _parse_qb_error(d.pop("qb_error", UNSET)) + + execute_event_block_response = cls( + event_id=event_id, + status=status, + qb_external_id=qb_external_id, + qb_error=qb_error, + ) + + execute_event_block_response.additional_properties = d + return execute_event_block_response + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/execute_event_block_response_qb_error_type_0.py b/robosystems_client/models/execute_event_block_response_qb_error_type_0.py new file mode 100644 index 0000000..9f0f65b --- /dev/null +++ b/robosystems_client/models/execute_event_block_response_qb_error_type_0.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, TypeVar + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +T = TypeVar("T", bound="ExecuteEventBlockResponseQbErrorType0") + + +@_attrs_define +class ExecuteEventBlockResponseQbErrorType0: + """ """ + + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + execute_event_block_response_qb_error_type_0 = cls() + + execute_event_block_response_qb_error_type_0.additional_properties = d + return execute_event_block_response_qb_error_type_0 + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/journal_entry_line_item_input.py b/robosystems_client/models/journal_entry_line_item_input.py index e63951f..2266596 100644 --- a/robosystems_client/models/journal_entry_line_item_input.py +++ b/robosystems_client/models/journal_entry_line_item_input.py @@ -1,13 +1,19 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Any, TypeVar, cast +from typing import TYPE_CHECKING, Any, TypeVar, cast from attrs import define as _attrs_define from attrs import field as _attrs_field from ..types import UNSET, Unset +if TYPE_CHECKING: + from ..models.journal_entry_line_item_input_metadata_type_0 import ( + JournalEntryLineItemInputMetadataType0, + ) + + T = TypeVar("T", bound="JournalEntryLineItemInput") @@ -23,15 +29,25 @@ class JournalEntryLineItemInput: debit_amount (int | Unset): Debit amount in cents. Must be 0 if `credit_amount` > 0. Default: 0. credit_amount (int | Unset): Credit amount in cents. Must be 0 if `debit_amount` > 0. Default: 0. description (None | str | Unset): Per-line memo (overrides the entry-level memo on this line). + metadata (JournalEntryLineItemInputMetadataType0 | None | Unset): Optional per-line metadata stamped on + ``LineItem.metadata_``. Used to carry source-system fields the standard columns don't cover — e.g. an external + flow-tag code that drives rollforward attribution (``transaction_description_code``), an external memo, or a + cost-center hint. Pass-through is non-validating; the renderer / filter engine reads keys it knows about and + ignores the rest. ``None`` is normalized to ``{}`` at persist time. """ element_id: str debit_amount: int | Unset = 0 credit_amount: int | Unset = 0 description: None | str | Unset = UNSET + metadata: JournalEntryLineItemInputMetadataType0 | None | Unset = UNSET additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) def to_dict(self) -> dict[str, Any]: + from ..models.journal_entry_line_item_input_metadata_type_0 import ( + JournalEntryLineItemInputMetadataType0, + ) + element_id = self.element_id debit_amount = self.debit_amount @@ -44,6 +60,14 @@ def to_dict(self) -> dict[str, Any]: else: description = self.description + metadata: dict[str, Any] | None | Unset + if isinstance(self.metadata, Unset): + metadata = UNSET + elif isinstance(self.metadata, JournalEntryLineItemInputMetadataType0): + metadata = self.metadata.to_dict() + else: + metadata = self.metadata + field_dict: dict[str, Any] = {} field_dict.update(self.additional_properties) field_dict.update( @@ -57,11 +81,17 @@ def to_dict(self) -> dict[str, Any]: field_dict["credit_amount"] = credit_amount if description is not UNSET: field_dict["description"] = description + if metadata is not UNSET: + field_dict["metadata"] = metadata return field_dict @classmethod def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.journal_entry_line_item_input_metadata_type_0 import ( + JournalEntryLineItemInputMetadataType0, + ) + d = dict(src_dict) element_id = d.pop("element_id") @@ -78,11 +108,31 @@ def _parse_description(data: object) -> None | str | Unset: description = _parse_description(d.pop("description", UNSET)) + def _parse_metadata( + data: object, + ) -> JournalEntryLineItemInputMetadataType0 | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, dict): + raise TypeError() + metadata_type_0 = JournalEntryLineItemInputMetadataType0.from_dict(data) + + return metadata_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(JournalEntryLineItemInputMetadataType0 | None | Unset, data) + + metadata = _parse_metadata(d.pop("metadata", UNSET)) + journal_entry_line_item_input = cls( element_id=element_id, debit_amount=debit_amount, credit_amount=credit_amount, description=description, + metadata=metadata, ) journal_entry_line_item_input.additional_properties = d diff --git a/robosystems_client/models/journal_entry_line_item_input_metadata_type_0.py b/robosystems_client/models/journal_entry_line_item_input_metadata_type_0.py new file mode 100644 index 0000000..fce2d53 --- /dev/null +++ b/robosystems_client/models/journal_entry_line_item_input_metadata_type_0.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, TypeVar + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +T = TypeVar("T", bound="JournalEntryLineItemInputMetadataType0") + + +@_attrs_define +class JournalEntryLineItemInputMetadataType0: + """ """ + + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + journal_entry_line_item_input_metadata_type_0 = cls() + + journal_entry_line_item_input_metadata_type_0.additional_properties = d + return journal_entry_line_item_input_metadata_type_0 + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/line_item_metadata_predicate.py b/robosystems_client/models/line_item_metadata_predicate.py new file mode 100644 index 0000000..ef7c1b8 --- /dev/null +++ b/robosystems_client/models/line_item_metadata_predicate.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, Literal, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..types import UNSET, Unset + +T = TypeVar("T", bound="LineItemMetadataPredicate") + + +@_attrs_define +class LineItemMetadataPredicate: + """Filter ledger LineItems whose ``metadata_[field]`` is in ``values``. + + The single predicate kind shipped in Phase 2 MVP. Sufficient for any + source taxonomy that stamps a flow-tag column on each transaction + line — mini's ``TransactionDescriptionCode``, future XBRL GL + ``GenericFlowCategory`` columns, custom tenant tags. + + ``field`` is the JSONB key under ``line_items.metadata`` (e.g. + ``"transaction_description_code"``). ``values`` is the set of values + that route to the filter's target concept; matched LineItems aggregate + signed into the attributed fact for the period. + + Attributes: + field (str): JSONB key under ``line_items.metadata`` to match against — e.g. ``transaction_description_code``. + The renderer performs an exact-string comparison on the JSONB-extracted text value. + values (list[str]): Metadata values that route to this filter's target concept. A LineItem matches when + ``metadata[field] ∈ values`` AND the line falls within the rollforward's period. + kind (Literal['line_item_metadata_field'] | Unset): Discriminator value selecting this predicate shape. Default: + 'line_item_metadata_field'. + """ + + field: str + values: list[str] + kind: Literal["line_item_metadata_field"] | Unset = "line_item_metadata_field" + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + field = self.field + + values = self.values + + kind = self.kind + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "field": field, + "values": values, + } + ) + if kind is not UNSET: + field_dict["kind"] = kind + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + field = d.pop("field") + + values = cast(list[str], d.pop("values")) + + kind = cast(Literal["line_item_metadata_field"] | Unset, d.pop("kind", UNSET)) + if kind != "line_item_metadata_field" and not isinstance(kind, Unset): + raise ValueError( + f"kind must match const 'line_item_metadata_field', got '{kind}'" + ) + + line_item_metadata_predicate = cls( + field=field, + values=values, + kind=kind, + ) + + line_item_metadata_predicate.additional_properties = d + return line_item_metadata_predicate + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/operation_envelope_execute_event_block_response.py b/robosystems_client/models/operation_envelope_execute_event_block_response.py new file mode 100644 index 0000000..ade4273 --- /dev/null +++ b/robosystems_client/models/operation_envelope_execute_event_block_response.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..models.operation_envelope_execute_event_block_response_status import ( + OperationEnvelopeExecuteEventBlockResponseStatus, +) +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.execute_event_block_response import ExecuteEventBlockResponse + + +T = TypeVar("T", bound="OperationEnvelopeExecuteEventBlockResponse") + + +@_attrs_define +class OperationEnvelopeExecuteEventBlockResponse: + """ + Attributes: + operation (str): Kebab-case operation name + operation_id (str): op_-prefixed ULID for audit and SSE correlation + status (OperationEnvelopeExecuteEventBlockResponseStatus): Operation lifecycle state + at (str): ISO-8601 UTC timestamp + result (ExecuteEventBlockResponse | None | Unset): Command-specific result payload + created_by (None | str | Unset): User ID that initiated the operation (null for legacy callers) + idempotent_replay (bool | Unset): True when this envelope came from the idempotency cache — the underlying + command did not execute again. False on fresh executions. Default: False. + """ + + operation: str + operation_id: str + status: OperationEnvelopeExecuteEventBlockResponseStatus + at: str + result: ExecuteEventBlockResponse | None | Unset = UNSET + created_by: None | str | Unset = UNSET + idempotent_replay: bool | Unset = False + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + from ..models.execute_event_block_response import ExecuteEventBlockResponse + + operation = self.operation + + operation_id = self.operation_id + + status = self.status.value + + at = self.at + + result: dict[str, Any] | None | Unset + if isinstance(self.result, Unset): + result = UNSET + elif isinstance(self.result, ExecuteEventBlockResponse): + result = self.result.to_dict() + else: + result = self.result + + created_by: None | str | Unset + if isinstance(self.created_by, Unset): + created_by = UNSET + else: + created_by = self.created_by + + idempotent_replay = self.idempotent_replay + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "operation": operation, + "operationId": operation_id, + "status": status, + "at": at, + } + ) + if result is not UNSET: + field_dict["result"] = result + if created_by is not UNSET: + field_dict["createdBy"] = created_by + if idempotent_replay is not UNSET: + field_dict["idempotentReplay"] = idempotent_replay + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.execute_event_block_response import ExecuteEventBlockResponse + + d = dict(src_dict) + operation = d.pop("operation") + + operation_id = d.pop("operationId") + + status = OperationEnvelopeExecuteEventBlockResponseStatus(d.pop("status")) + + at = d.pop("at") + + def _parse_result(data: object) -> ExecuteEventBlockResponse | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, dict): + raise TypeError() + result_type_0 = ExecuteEventBlockResponse.from_dict(data) + + return result_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(ExecuteEventBlockResponse | None | Unset, data) + + result = _parse_result(d.pop("result", UNSET)) + + def _parse_created_by(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + created_by = _parse_created_by(d.pop("createdBy", UNSET)) + + idempotent_replay = d.pop("idempotentReplay", UNSET) + + operation_envelope_execute_event_block_response = cls( + operation=operation, + operation_id=operation_id, + status=status, + at=at, + result=result, + created_by=created_by, + idempotent_replay=idempotent_replay, + ) + + operation_envelope_execute_event_block_response.additional_properties = d + return operation_envelope_execute_event_block_response + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/operation_envelope_execute_event_block_response_status.py b/robosystems_client/models/operation_envelope_execute_event_block_response_status.py new file mode 100644 index 0000000..f82c40f --- /dev/null +++ b/robosystems_client/models/operation_envelope_execute_event_block_response_status.py @@ -0,0 +1,10 @@ +from enum import Enum + + +class OperationEnvelopeExecuteEventBlockResponseStatus(str, Enum): + COMPLETED = "completed" + FAILED = "failed" + PENDING = "pending" + + def __str__(self) -> str: + return str(self.value) diff --git a/robosystems_client/models/rollforward_mechanics.py b/robosystems_client/models/rollforward_mechanics.py new file mode 100644 index 0000000..fbf6bdd --- /dev/null +++ b/robosystems_client/models/rollforward_mechanics.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..models.rollforward_mechanics_validation_mode import ( + RollforwardMechanicsValidationMode, +) +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.attribution_filter import AttributionFilter + + +T = TypeVar("T", bound="RollforwardMechanics") + + +@_attrs_define +class RollforwardMechanics: + """Filter-based attribution mechanics for ``block_type='rollforward'``. + + Implements Tier 2 of the rollforward attribution design + (``information-block.md`` §4.5). Each block decomposes one BS + source element's period delta into a list of flow concepts via + declared :class:`AttributionFilter` predicates. The renderer + evaluates the filters against ledger LineItems at envelope-build + time, emits one attributed fact per filter per period, and arbitrates + any residual against the default change tag (Tier 1 fallback). + + Reads directly from the typed ``structures.artifact_mechanics`` JSONB + column. ``attribution_filters`` rides as nested JSON; the predicate + union widens as new predicate shapes ship (Phase 2 MVP carries only + ``line_item_metadata_field``). + + Attributes: + bs_source_element_id (str): Element id of the balance-sheet source whose period delta this block decomposes. + Resolved from ``bs_source_qname`` at create time. + bs_source_qname (str): QName of the BS source element (e.g. ``mini:CashAndCashEquivalents``). Round-tripped for + caller convenience; ``bs_source_element_id`` is authoritative. + kind (Literal['rollforward'] | Unset): Default: 'rollforward'. + default_change_tag_element_id (None | str | Unset): Element id of the Tier 1 default change tag — the fallback + flow concept that receives any residual (Δ BS − Σ filter matches). Null when no default is declared; behavior on + residual then follows ``validation_mode``. + default_change_tag_qname (None | str | Unset): QName of the Tier 1 default change tag (e.g. ``rs- + gaap:IncreaseDecreaseInCashAndCashEquivalents``). Round-tripped for caller convenience and operator-readable + envelopes; ``default_change_tag_element_id`` is authoritative. Null iff ``default_change_tag_element_id`` is + null. + attribution_filters (list[AttributionFilter] | Unset): Filter predicates routing LineItems to flow concepts. The + renderer evaluates each filter against the period's LineItems, aggregates signed amounts, and emits one fact per + filter per period. + validation_mode (RollforwardMechanicsValidationMode | Unset): Renderer arbitration policy when Σ filter matches + != Δ BS. ``strict`` raises; ``residual_as_default`` emits the residual as a default-tag fact (the common case); + ``warn_only`` logs and lets the imbalance pass. Default: RollforwardMechanicsValidationMode.RESIDUAL_AS_DEFAULT. + """ + + bs_source_element_id: str + bs_source_qname: str + kind: Literal["rollforward"] | Unset = "rollforward" + default_change_tag_element_id: None | str | Unset = UNSET + default_change_tag_qname: None | str | Unset = UNSET + attribution_filters: list[AttributionFilter] | Unset = UNSET + validation_mode: RollforwardMechanicsValidationMode | Unset = ( + RollforwardMechanicsValidationMode.RESIDUAL_AS_DEFAULT + ) + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + bs_source_element_id = self.bs_source_element_id + + bs_source_qname = self.bs_source_qname + + kind = self.kind + + default_change_tag_element_id: None | str | Unset + if isinstance(self.default_change_tag_element_id, Unset): + default_change_tag_element_id = UNSET + else: + default_change_tag_element_id = self.default_change_tag_element_id + + default_change_tag_qname: None | str | Unset + if isinstance(self.default_change_tag_qname, Unset): + default_change_tag_qname = UNSET + else: + default_change_tag_qname = self.default_change_tag_qname + + attribution_filters: list[dict[str, Any]] | Unset = UNSET + if not isinstance(self.attribution_filters, Unset): + attribution_filters = [] + for attribution_filters_item_data in self.attribution_filters: + attribution_filters_item = attribution_filters_item_data.to_dict() + attribution_filters.append(attribution_filters_item) + + validation_mode: str | Unset = UNSET + if not isinstance(self.validation_mode, Unset): + validation_mode = self.validation_mode.value + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "bs_source_element_id": bs_source_element_id, + "bs_source_qname": bs_source_qname, + } + ) + if kind is not UNSET: + field_dict["kind"] = kind + if default_change_tag_element_id is not UNSET: + field_dict["default_change_tag_element_id"] = default_change_tag_element_id + if default_change_tag_qname is not UNSET: + field_dict["default_change_tag_qname"] = default_change_tag_qname + if attribution_filters is not UNSET: + field_dict["attribution_filters"] = attribution_filters + if validation_mode is not UNSET: + field_dict["validation_mode"] = validation_mode + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.attribution_filter import AttributionFilter + + d = dict(src_dict) + bs_source_element_id = d.pop("bs_source_element_id") + + bs_source_qname = d.pop("bs_source_qname") + + kind = cast(Literal["rollforward"] | Unset, d.pop("kind", UNSET)) + if kind != "rollforward" and not isinstance(kind, Unset): + raise ValueError(f"kind must match const 'rollforward', got '{kind}'") + + def _parse_default_change_tag_element_id(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + default_change_tag_element_id = _parse_default_change_tag_element_id( + d.pop("default_change_tag_element_id", UNSET) + ) + + def _parse_default_change_tag_qname(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + default_change_tag_qname = _parse_default_change_tag_qname( + d.pop("default_change_tag_qname", UNSET) + ) + + _attribution_filters = d.pop("attribution_filters", UNSET) + attribution_filters: list[AttributionFilter] | Unset = UNSET + if _attribution_filters is not UNSET: + attribution_filters = [] + for attribution_filters_item_data in _attribution_filters: + attribution_filters_item = AttributionFilter.from_dict( + attribution_filters_item_data + ) + + attribution_filters.append(attribution_filters_item) + + _validation_mode = d.pop("validation_mode", UNSET) + validation_mode: RollforwardMechanicsValidationMode | Unset + if isinstance(_validation_mode, Unset): + validation_mode = UNSET + else: + validation_mode = RollforwardMechanicsValidationMode(_validation_mode) + + rollforward_mechanics = cls( + bs_source_element_id=bs_source_element_id, + bs_source_qname=bs_source_qname, + kind=kind, + default_change_tag_element_id=default_change_tag_element_id, + default_change_tag_qname=default_change_tag_qname, + attribution_filters=attribution_filters, + validation_mode=validation_mode, + ) + + rollforward_mechanics.additional_properties = d + return rollforward_mechanics + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/rollforward_mechanics_validation_mode.py b/robosystems_client/models/rollforward_mechanics_validation_mode.py new file mode 100644 index 0000000..ee97531 --- /dev/null +++ b/robosystems_client/models/rollforward_mechanics_validation_mode.py @@ -0,0 +1,10 @@ +from enum import Enum + + +class RollforwardMechanicsValidationMode(str, Enum): + RESIDUAL_AS_DEFAULT = "residual_as_default" + STRICT = "strict" + WARN_ONLY = "warn_only" + + def __str__(self) -> str: + return str(self.value) diff --git a/robosystems_client/models/update_rollforward_arm.py b/robosystems_client/models/update_rollforward_arm.py new file mode 100644 index 0000000..6b28f2f --- /dev/null +++ b/robosystems_client/models/update_rollforward_arm.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +if TYPE_CHECKING: + from ..models.update_rollforward_request import UpdateRollforwardRequest + + +T = TypeVar("T", bound="UpdateRollforwardArm") + + +@_attrs_define +class UpdateRollforwardArm: + """Update-information-block body for ``block_type="rollforward"``. + + Carries a typed rollforward update payload. Mutable fields: name, + default_change_tag_qname, attribution_filters, validation_mode. + + Attributes: + block_type (Literal['rollforward']): Discriminator value selecting this arm. + payload (UpdateRollforwardRequest): Update mutable fields on a rollforward block. + + Editable: name, default_change_tag_qname, attribution_filters, + validation_mode. The BS source is fixed once the block is created + (changing it would invalidate every previously rendered period); to + change BS source, delete and re-create. + + **Partial-update semantics**: omitted (``None``) fields mean "leave + unchanged" — there is no wire-level way to *clear* a previously set + default change tag or empty the attribution_filters list via this + endpoint. To remove the default tag entirely, delete and re-create + the rollforward block. The asymmetry is deliberate: an explicit + clear-sentinel adds wire-shape complexity for a use case that rarely + arises in practice (default tags are typically set during initial + authoring and only swapped, not removed). + """ + + block_type: Literal["rollforward"] + payload: UpdateRollforwardRequest + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + block_type = self.block_type + + payload = self.payload.to_dict() + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "block_type": block_type, + "payload": payload, + } + ) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.update_rollforward_request import UpdateRollforwardRequest + + d = dict(src_dict) + block_type = cast(Literal["rollforward"], d.pop("block_type")) + if block_type != "rollforward": + raise ValueError(f"block_type must match const 'rollforward', got '{block_type}'") + + payload = UpdateRollforwardRequest.from_dict(d.pop("payload")) + + update_rollforward_arm = cls( + block_type=block_type, + payload=payload, + ) + + update_rollforward_arm.additional_properties = d + return update_rollforward_arm + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/update_rollforward_request.py b/robosystems_client/models/update_rollforward_request.py new file mode 100644 index 0000000..a01fd22 --- /dev/null +++ b/robosystems_client/models/update_rollforward_request.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +from ..models.update_rollforward_request_validation_mode_type_0 import ( + UpdateRollforwardRequestValidationModeType0, +) +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.attribution_filter import AttributionFilter + + +T = TypeVar("T", bound="UpdateRollforwardRequest") + + +@_attrs_define +class UpdateRollforwardRequest: + """Update mutable fields on a rollforward block. + + Editable: name, default_change_tag_qname, attribution_filters, + validation_mode. The BS source is fixed once the block is created + (changing it would invalidate every previously rendered period); to + change BS source, delete and re-create. + + **Partial-update semantics**: omitted (``None``) fields mean "leave + unchanged" — there is no wire-level way to *clear* a previously set + default change tag or empty the attribution_filters list via this + endpoint. To remove the default tag entirely, delete and re-create + the rollforward block. The asymmetry is deliberate: an explicit + clear-sentinel adds wire-shape complexity for a use case that rarely + arises in practice (default tags are typically set during initial + authoring and only swapped, not removed). + + Attributes: + structure_id (str): Structure ID of the rollforward block. + name (None | str | Unset): + default_change_tag_qname (None | str | Unset): New default change tag qname. Pass a value to *change* the + default; omit (``None``) to leave unchanged. There is no wire-level way to clear a previously set default — see + the class docstring. + attribution_filters (list[AttributionFilter] | None | Unset): + validation_mode (None | Unset | UpdateRollforwardRequestValidationModeType0): + """ + + structure_id: str + name: None | str | Unset = UNSET + default_change_tag_qname: None | str | Unset = UNSET + attribution_filters: list[AttributionFilter] | None | Unset = UNSET + validation_mode: None | Unset | UpdateRollforwardRequestValidationModeType0 = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + structure_id = self.structure_id + + name: None | str | Unset + if isinstance(self.name, Unset): + name = UNSET + else: + name = self.name + + default_change_tag_qname: None | str | Unset + if isinstance(self.default_change_tag_qname, Unset): + default_change_tag_qname = UNSET + else: + default_change_tag_qname = self.default_change_tag_qname + + attribution_filters: list[dict[str, Any]] | None | Unset + if isinstance(self.attribution_filters, Unset): + attribution_filters = UNSET + elif isinstance(self.attribution_filters, list): + attribution_filters = [] + for attribution_filters_type_0_item_data in self.attribution_filters: + attribution_filters_type_0_item = attribution_filters_type_0_item_data.to_dict() + attribution_filters.append(attribution_filters_type_0_item) + + else: + attribution_filters = self.attribution_filters + + validation_mode: None | str | Unset + if isinstance(self.validation_mode, Unset): + validation_mode = UNSET + elif isinstance(self.validation_mode, UpdateRollforwardRequestValidationModeType0): + validation_mode = self.validation_mode.value + else: + validation_mode = self.validation_mode + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "structure_id": structure_id, + } + ) + if name is not UNSET: + field_dict["name"] = name + if default_change_tag_qname is not UNSET: + field_dict["default_change_tag_qname"] = default_change_tag_qname + if attribution_filters is not UNSET: + field_dict["attribution_filters"] = attribution_filters + if validation_mode is not UNSET: + field_dict["validation_mode"] = validation_mode + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.attribution_filter import AttributionFilter + + d = dict(src_dict) + structure_id = d.pop("structure_id") + + def _parse_name(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + name = _parse_name(d.pop("name", UNSET)) + + def _parse_default_change_tag_qname(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + default_change_tag_qname = _parse_default_change_tag_qname( + d.pop("default_change_tag_qname", UNSET) + ) + + def _parse_attribution_filters( + data: object, + ) -> list[AttributionFilter] | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, list): + raise TypeError() + attribution_filters_type_0 = [] + _attribution_filters_type_0 = data + for attribution_filters_type_0_item_data in _attribution_filters_type_0: + attribution_filters_type_0_item = AttributionFilter.from_dict( + attribution_filters_type_0_item_data + ) + + attribution_filters_type_0.append(attribution_filters_type_0_item) + + return attribution_filters_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(list[AttributionFilter] | None | Unset, data) + + attribution_filters = _parse_attribution_filters( + d.pop("attribution_filters", UNSET) + ) + + def _parse_validation_mode( + data: object, + ) -> None | Unset | UpdateRollforwardRequestValidationModeType0: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, str): + raise TypeError() + validation_mode_type_0 = UpdateRollforwardRequestValidationModeType0(data) + + return validation_mode_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(None | Unset | UpdateRollforwardRequestValidationModeType0, data) + + validation_mode = _parse_validation_mode(d.pop("validation_mode", UNSET)) + + update_rollforward_request = cls( + structure_id=structure_id, + name=name, + default_change_tag_qname=default_change_tag_qname, + attribution_filters=attribution_filters, + validation_mode=validation_mode, + ) + + update_rollforward_request.additional_properties = d + return update_rollforward_request + + @property + def additional_keys(self) -> list[str]: + return list(self.additional_properties.keys()) + + def __getitem__(self, key: str) -> Any: + return self.additional_properties[key] + + def __setitem__(self, key: str, value: Any) -> None: + self.additional_properties[key] = value + + def __delitem__(self, key: str) -> None: + del self.additional_properties[key] + + def __contains__(self, key: str) -> bool: + return key in self.additional_properties diff --git a/robosystems_client/models/update_rollforward_request_validation_mode_type_0.py b/robosystems_client/models/update_rollforward_request_validation_mode_type_0.py new file mode 100644 index 0000000..aeb9cad --- /dev/null +++ b/robosystems_client/models/update_rollforward_request_validation_mode_type_0.py @@ -0,0 +1,10 @@ +from enum import Enum + + +class UpdateRollforwardRequestValidationModeType0(str, Enum): + RESIDUAL_AS_DEFAULT = "residual_as_default" + STRICT = "strict" + WARN_ONLY = "warn_only" + + def __str__(self) -> str: + return str(self.value)