diff --git a/robosystems_client/api/robo_ledger_analytical_views/consolidation_worksheet.py b/robosystems_client/api/robo_ledger_analytical_views/consolidation_worksheet.py new file mode 100644 index 0000000..bcac6c9 --- /dev/null +++ b/robosystems_client/api/robo_ledger_analytical_views/consolidation_worksheet.py @@ -0,0 +1,276 @@ +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.consolidation_worksheet_request import ConsolidationWorksheetRequest +from ...models.error_response import ErrorResponse +from ...models.operation_envelope_consolidation_worksheet_response import ( + OperationEnvelopeConsolidationWorksheetResponse, +) +from ...types import UNSET, Response, Unset + + +def _get_kwargs( + graph_id: str, + *, + body: ConsolidationWorksheetRequest, + 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/consolidation-worksheet".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 | OperationEnvelopeConsolidationWorksheetResponse | None: + if response.status_code == 200: + response_200 = OperationEnvelopeConsolidationWorksheetResponse.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 | OperationEnvelopeConsolidationWorksheetResponse]: + 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: ConsolidationWorksheetRequest, + idempotency_key: None | str | Unset = UNSET, +) -> Response[ErrorResponse | OperationEnvelopeConsolidationWorksheetResponse]: + """Consolidation Worksheet + + The group's statement for one window with the entity as the column, read live from the tenant's OLTP + ledger: the parent and each member consolidated in full, their combined sum, the consolidation + entity's adjustments, the consolidated total, and each equity-method or cost member beside the sum, + never in it. Every column renders under the group parent's Reporting Style. Rejected on shared- + repository graphs. + + **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 (ConsolidationWorksheetRequest): Request for consolidation-worksheet (OLTP, entity + graphs only). + + 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 | OperationEnvelopeConsolidationWorksheetResponse] + """ + + 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: ConsolidationWorksheetRequest, + idempotency_key: None | str | Unset = UNSET, +) -> ErrorResponse | OperationEnvelopeConsolidationWorksheetResponse | None: + """Consolidation Worksheet + + The group's statement for one window with the entity as the column, read live from the tenant's OLTP + ledger: the parent and each member consolidated in full, their combined sum, the consolidation + entity's adjustments, the consolidated total, and each equity-method or cost member beside the sum, + never in it. Every column renders under the group parent's Reporting Style. Rejected on shared- + repository graphs. + + **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 (ConsolidationWorksheetRequest): Request for consolidation-worksheet (OLTP, entity + graphs only). + + 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 | OperationEnvelopeConsolidationWorksheetResponse + """ + + 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: ConsolidationWorksheetRequest, + idempotency_key: None | str | Unset = UNSET, +) -> Response[ErrorResponse | OperationEnvelopeConsolidationWorksheetResponse]: + """Consolidation Worksheet + + The group's statement for one window with the entity as the column, read live from the tenant's OLTP + ledger: the parent and each member consolidated in full, their combined sum, the consolidation + entity's adjustments, the consolidated total, and each equity-method or cost member beside the sum, + never in it. Every column renders under the group parent's Reporting Style. Rejected on shared- + repository graphs. + + **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 (ConsolidationWorksheetRequest): Request for consolidation-worksheet (OLTP, entity + graphs only). + + 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 | OperationEnvelopeConsolidationWorksheetResponse] + """ + + 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: ConsolidationWorksheetRequest, + idempotency_key: None | str | Unset = UNSET, +) -> ErrorResponse | OperationEnvelopeConsolidationWorksheetResponse | None: + """Consolidation Worksheet + + The group's statement for one window with the entity as the column, read live from the tenant's OLTP + ledger: the parent and each member consolidated in full, their combined sum, the consolidation + entity's adjustments, the consolidated total, and each equity-method or cost member beside the sum, + never in it. Every column renders under the group parent's Reporting Style. Rejected on shared- + repository graphs. + + **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 (ConsolidationWorksheetRequest): Request for consolidation-worksheet (OLTP, entity + graphs only). + + 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 | OperationEnvelopeConsolidationWorksheetResponse + """ + + return ( + await asyncio_detailed( + graph_id=graph_id, + client=client, + body=body, + idempotency_key=idempotency_key, + ) + ).parsed diff --git a/robosystems_client/clients/ledger_client.py b/robosystems_client/clients/ledger_client.py index 2a12b93..f4c6f47 100644 --- a/robosystems_client/clients/ledger_client.py +++ b/robosystems_client/clients/ledger_client.py @@ -81,6 +81,9 @@ from ..api.robo_ledger_analytical_views.financial_statement_analysis import ( sync_detailed as op_financial_statement_analysis, ) +from ..api.robo_ledger_analytical_views.consolidation_worksheet import ( + sync_detailed as op_consolidation_worksheet, +) from ..api.robo_ledger_analytical_views.live_financial_statement import ( sync_detailed as op_live_financial_statement, ) @@ -519,6 +522,8 @@ from ..models.financial_statement_analysis_request import ( FinancialStatementAnalysisRequest, ) +from ..models.consolidation_worksheet_request import ConsolidationWorksheetRequest +from ..models.consolidation_worksheet_response import ConsolidationWorksheetResponse from ..models.live_financial_statement_request import LiveFinancialStatementRequest from ..models.financial_statement_analysis_response import ( FinancialStatementAnalysisResponse, @@ -2346,6 +2351,25 @@ def live_financial_statement( "Live financial statement", envelope, LiveFinancialStatementResponse ) + def consolidation_worksheet( + self, + graph_id: str, + body: dict[str, Any], + ) -> ConsolidationWorksheetResponse: + """The group's statement for one window with the entity as the column: + each member consolidated in full, their combined sum, the consolidation + entity's adjustments, the consolidated total, and each equity-method or + cost member beside the sum. Read on the group parent. + """ + request = ConsolidationWorksheetRequest.from_dict(body) + response = op_consolidation_worksheet( + graph_id=graph_id, body=request, client=self._get_client() + ) + envelope = self._call_op("Consolidation worksheet", response) + return self._typed_result( + "Consolidation worksheet", envelope, ConsolidationWorksheetResponse + ) + def financial_statement_analysis( self, graph_id: str, diff --git a/robosystems_client/models/__init__.py b/robosystems_client/models/__init__.py index 1514cdb..7b7360b 100644 --- a/robosystems_client/models/__init__.py +++ b/robosystems_client/models/__init__.py @@ -91,6 +91,8 @@ from .connection_provider_info_provider import ConnectionProviderInfoProvider from .connection_response import ConnectionResponse from .connection_response_metadata import ConnectionResponseMetadata +from .consolidation_worksheet_request import ConsolidationWorksheetRequest +from .consolidation_worksheet_response import ConsolidationWorksheetResponse from .content_limits import ContentLimits from .copy_operation_limits import CopyOperationLimits from .create_agent_request import CreateAgentRequest @@ -533,6 +535,12 @@ from .operation_envelope_compute_metrics_response_status import ( OperationEnvelopeComputeMetricsResponseStatus, ) +from .operation_envelope_consolidation_worksheet_response import ( + OperationEnvelopeConsolidationWorksheetResponse, +) +from .operation_envelope_consolidation_worksheet_response_status import ( + OperationEnvelopeConsolidationWorksheetResponseStatus, +) from .operation_envelope_delete_information_block_response import ( OperationEnvelopeDeleteInformationBlockResponse, ) @@ -1169,6 +1177,9 @@ from .view_projections import ViewProjections from .view_response import ViewResponse from .view_response_summary_type_0 import ViewResponseSummaryType0 +from .worksheet_column import WorksheetColumn +from .worksheet_column_kind import WorksheetColumnKind +from .worksheet_row import WorksheetRow __all__ = ( "AcceptSuggestionLine", @@ -1250,6 +1261,8 @@ "ConnectionProviderInfoProvider", "ConnectionResponse", "ConnectionResponseMetadata", + "ConsolidationWorksheetRequest", + "ConsolidationWorksheetResponse", "ContentLimits", "CopyOperationLimits", "CreateAgentRequest", @@ -1566,6 +1579,8 @@ "OperationEnvelopeComputeForecastResponseStatus", "OperationEnvelopeComputeMetricsResponse", "OperationEnvelopeComputeMetricsResponseStatus", + "OperationEnvelopeConsolidationWorksheetResponse", + "OperationEnvelopeConsolidationWorksheetResponseStatus", "OperationEnvelopeDeleteInformationBlockResponse", "OperationEnvelopeDeleteInformationBlockResponseStatus", "OperationEnvelopeDeletePortfolioBlockResponse", @@ -1956,4 +1971,7 @@ "ViewProjections", "ViewResponse", "ViewResponseSummaryType0", + "WorksheetColumn", + "WorksheetColumnKind", + "WorksheetRow", ) diff --git a/robosystems_client/models/consolidation_worksheet_request.py b/robosystems_client/models/consolidation_worksheet_request.py new file mode 100644 index 0000000..723293b --- /dev/null +++ b/robosystems_client/models/consolidation_worksheet_request.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +import datetime +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="ConsolidationWorksheetRequest") + + +@_attrs_define +class ConsolidationWorksheetRequest: + """Request for consolidation-worksheet (OLTP, entity graphs only). + + Attributes: + statement_type (str): income_statement | balance_sheet | cash_flow_statement | equity_statement. + period_start (datetime.date | None | Unset): Explicit window start. Overrides period_type/fiscal_year. + period_end (datetime.date | None | Unset): Explicit window end. Overrides period_type/fiscal_year. + period_type (None | str | Unset): annual | quarterly | instant (ignored when dates supplied) + fiscal_year (int | None | Unset): Fiscal year for annual window (anchored on FiscalCalendar) + limit (int | Unset): Max rows returned. Defaults to the ceiling: the whole statement. Default: 1000. + """ + + statement_type: str + period_start: datetime.date | None | Unset = UNSET + period_end: datetime.date | None | Unset = UNSET + period_type: None | str | Unset = UNSET + fiscal_year: int | None | Unset = UNSET + limit: int | Unset = 1000 + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + statement_type = self.statement_type + + period_start: None | str | Unset + if isinstance(self.period_start, Unset): + period_start = UNSET + elif isinstance(self.period_start, datetime.date): + period_start = self.period_start.isoformat() + else: + period_start = self.period_start + + period_end: None | str | Unset + if isinstance(self.period_end, Unset): + period_end = UNSET + elif isinstance(self.period_end, datetime.date): + period_end = self.period_end.isoformat() + else: + period_end = self.period_end + + period_type: None | str | Unset + if isinstance(self.period_type, Unset): + period_type = UNSET + else: + period_type = self.period_type + + fiscal_year: int | None | Unset + if isinstance(self.fiscal_year, Unset): + fiscal_year = UNSET + else: + fiscal_year = self.fiscal_year + + limit = self.limit + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "statement_type": statement_type, + } + ) + if period_start is not UNSET: + field_dict["period_start"] = period_start + if period_end is not UNSET: + field_dict["period_end"] = period_end + if period_type is not UNSET: + field_dict["period_type"] = period_type + if fiscal_year is not UNSET: + field_dict["fiscal_year"] = fiscal_year + if limit is not UNSET: + field_dict["limit"] = limit + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + statement_type = d.pop("statement_type") + + def _parse_period_start(data: object) -> datetime.date | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, str): + raise TypeError() + period_start_type_0 = datetime.date.fromisoformat(data) + + return period_start_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(datetime.date | None | Unset, data) + + period_start = _parse_period_start(d.pop("period_start", UNSET)) + + def _parse_period_end(data: object) -> datetime.date | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, str): + raise TypeError() + period_end_type_0 = datetime.date.fromisoformat(data) + + return period_end_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(datetime.date | None | Unset, data) + + period_end = _parse_period_end(d.pop("period_end", UNSET)) + + def _parse_period_type(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + period_type = _parse_period_type(d.pop("period_type", UNSET)) + + def _parse_fiscal_year(data: object) -> int | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(int | None | Unset, data) + + fiscal_year = _parse_fiscal_year(d.pop("fiscal_year", UNSET)) + + limit = d.pop("limit", UNSET) + + consolidation_worksheet_request = cls( + statement_type=statement_type, + period_start=period_start, + period_end=period_end, + period_type=period_type, + fiscal_year=fiscal_year, + limit=limit, + ) + + consolidation_worksheet_request.additional_properties = d + return consolidation_worksheet_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/consolidation_worksheet_response.py b/robosystems_client/models/consolidation_worksheet_response.py new file mode 100644 index 0000000..4f692c0 --- /dev/null +++ b/robosystems_client/models/consolidation_worksheet_response.py @@ -0,0 +1,203 @@ +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.period_spec import PeriodSpec + from ..models.validation_check_response import ValidationCheckResponse + from ..models.worksheet_column import WorksheetColumn + from ..models.worksheet_row import WorksheetRow + + +T = TypeVar("T", bound="ConsolidationWorksheetResponse") + + +@_attrs_define +class ConsolidationWorksheetResponse: + """The group's statement with the entity as the column. + + Attributes: + graph_id (str): The graph read. + entity_id (str): The group parent. + statement_type (str): The statement rendered. + period (PeriodSpec): A single reporting period column. + + Reports render facts in N period columns side-by-side. Each + `PeriodSpec` is one column — its `start`/`end` define the + window the report's facts roll up into; `label` is what the renderer + prints in the column header. For year-over-year statements, supply two + PeriodSpecs (current + comparative); for YTD by quarter, supply four. + columns (list[WorksheetColumn]): The parent, each member consolidated in full, `combined`, the consolidation + entity's adjustments (when the group has one), `consolidated`, then the equity-method and cost members outside + the sum. A member outside its membership interval for the period has no column. + rows (list[WorksheetRow]): The union of every column's lines in the parent's Style order; a line zero in every + column is left out. + validation (None | Unset | ValidationCheckResponse): The guard-rail outcome for the consolidated column, the + same checks live-financial-statement runs. + unmapped_entity_ids (list[str] | Unset): Entities with no CoA mapping yet, so with no column. + unmapped_count (int | Unset): Unmapped accounts across the summed columns. Default: 0. + truncated (bool | Unset): Whether `limit` cut rows. Default: False. + """ + + graph_id: str + entity_id: str + statement_type: str + period: PeriodSpec + columns: list[WorksheetColumn] + rows: list[WorksheetRow] + validation: None | Unset | ValidationCheckResponse = UNSET + unmapped_entity_ids: list[str] | Unset = UNSET + unmapped_count: int | Unset = 0 + truncated: bool | Unset = False + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + from ..models.validation_check_response import ValidationCheckResponse + + graph_id = self.graph_id + + entity_id = self.entity_id + + statement_type = self.statement_type + + period = self.period.to_dict() + + columns = [] + for columns_item_data in self.columns: + columns_item = columns_item_data.to_dict() + columns.append(columns_item) + + rows = [] + for rows_item_data in self.rows: + rows_item = rows_item_data.to_dict() + rows.append(rows_item) + + validation: dict[str, Any] | None | Unset + if isinstance(self.validation, Unset): + validation = UNSET + elif isinstance(self.validation, ValidationCheckResponse): + validation = self.validation.to_dict() + else: + validation = self.validation + + unmapped_entity_ids: list[str] | Unset = UNSET + if not isinstance(self.unmapped_entity_ids, Unset): + unmapped_entity_ids = self.unmapped_entity_ids + + unmapped_count = self.unmapped_count + + truncated = self.truncated + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "graph_id": graph_id, + "entity_id": entity_id, + "statement_type": statement_type, + "period": period, + "columns": columns, + "rows": rows, + } + ) + if validation is not UNSET: + field_dict["validation"] = validation + if unmapped_entity_ids is not UNSET: + field_dict["unmapped_entity_ids"] = unmapped_entity_ids + if unmapped_count is not UNSET: + field_dict["unmapped_count"] = unmapped_count + if truncated is not UNSET: + field_dict["truncated"] = truncated + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.period_spec import PeriodSpec + from ..models.validation_check_response import ValidationCheckResponse + from ..models.worksheet_column import WorksheetColumn + from ..models.worksheet_row import WorksheetRow + + d = dict(src_dict) + graph_id = d.pop("graph_id") + + entity_id = d.pop("entity_id") + + statement_type = d.pop("statement_type") + + period = PeriodSpec.from_dict(d.pop("period")) + + columns = [] + _columns = d.pop("columns") + for columns_item_data in _columns: + columns_item = WorksheetColumn.from_dict(columns_item_data) + + columns.append(columns_item) + + rows = [] + _rows = d.pop("rows") + for rows_item_data in _rows: + rows_item = WorksheetRow.from_dict(rows_item_data) + + rows.append(rows_item) + + def _parse_validation(data: object) -> None | Unset | ValidationCheckResponse: + if data is None: + return data + if isinstance(data, Unset): + return data + try: + if not isinstance(data, dict): + raise TypeError() + validation_type_0 = ValidationCheckResponse.from_dict(data) + + return validation_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(None | Unset | ValidationCheckResponse, data) + + validation = _parse_validation(d.pop("validation", UNSET)) + + unmapped_entity_ids = cast(list[str], d.pop("unmapped_entity_ids", UNSET)) + + unmapped_count = d.pop("unmapped_count", UNSET) + + truncated = d.pop("truncated", UNSET) + + consolidation_worksheet_response = cls( + graph_id=graph_id, + entity_id=entity_id, + statement_type=statement_type, + period=period, + columns=columns, + rows=rows, + validation=validation, + unmapped_entity_ids=unmapped_entity_ids, + unmapped_count=unmapped_count, + truncated=truncated, + ) + + consolidation_worksheet_response.additional_properties = d + return consolidation_worksheet_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_consolidation_worksheet_response.py b/robosystems_client/models/operation_envelope_consolidation_worksheet_response.py new file mode 100644 index 0000000..19998fa --- /dev/null +++ b/robosystems_client/models/operation_envelope_consolidation_worksheet_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_consolidation_worksheet_response_status import ( + OperationEnvelopeConsolidationWorksheetResponseStatus, +) +from ..types import UNSET, Unset + +if TYPE_CHECKING: + from ..models.consolidation_worksheet_response import ConsolidationWorksheetResponse + + +T = TypeVar("T", bound="OperationEnvelopeConsolidationWorksheetResponse") + + +@_attrs_define +class OperationEnvelopeConsolidationWorksheetResponse: + """ + Attributes: + operation (str): Kebab-case operation name + operation_id (str): op_-prefixed ULID for audit and SSE correlation + status (OperationEnvelopeConsolidationWorksheetResponseStatus): Operation lifecycle state + at (str): ISO-8601 UTC timestamp + result (ConsolidationWorksheetResponse | None | Unset): Command-specific result payload + created_by (None | str | Unset): User ID that initiated the operation + 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: OperationEnvelopeConsolidationWorksheetResponseStatus + at: str + result: ConsolidationWorksheetResponse | 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.consolidation_worksheet_response import ConsolidationWorksheetResponse + + 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, ConsolidationWorksheetResponse): + 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.consolidation_worksheet_response import ConsolidationWorksheetResponse + + d = dict(src_dict) + operation = d.pop("operation") + + operation_id = d.pop("operationId") + + status = OperationEnvelopeConsolidationWorksheetResponseStatus(d.pop("status")) + + at = d.pop("at") + + def _parse_result(data: object) -> ConsolidationWorksheetResponse | 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 = ConsolidationWorksheetResponse.from_dict(data) + + return result_type_0 + except (TypeError, ValueError, AttributeError, KeyError): + pass + return cast(ConsolidationWorksheetResponse | 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_consolidation_worksheet_response = cls( + operation=operation, + operation_id=operation_id, + status=status, + at=at, + result=result, + created_by=created_by, + idempotent_replay=idempotent_replay, + ) + + operation_envelope_consolidation_worksheet_response.additional_properties = d + return operation_envelope_consolidation_worksheet_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_consolidation_worksheet_response_status.py b/robosystems_client/models/operation_envelope_consolidation_worksheet_response_status.py new file mode 100644 index 0000000..97b1f22 --- /dev/null +++ b/robosystems_client/models/operation_envelope_consolidation_worksheet_response_status.py @@ -0,0 +1,10 @@ +from enum import Enum + + +class OperationEnvelopeConsolidationWorksheetResponseStatus(str, Enum): + COMPLETED = "completed" + FAILED = "failed" + PENDING = "pending" + + def __str__(self) -> str: + return str(self.value) diff --git a/robosystems_client/models/worksheet_column.py b/robosystems_client/models/worksheet_column.py new file mode 100644 index 0000000..983403f --- /dev/null +++ b/robosystems_client/models/worksheet_column.py @@ -0,0 +1,153 @@ +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 ..models.worksheet_column_kind import WorksheetColumnKind +from ..types import UNSET, Unset + +T = TypeVar("T", bound="WorksheetColumn") + + +@_attrs_define +class WorksheetColumn: + """One column of the consolidation worksheet. + + Attributes: + key (str): The entity id for an entity's column; `combined` and `consolidated` for the two totals. + label (str): The entity's name, or the total's. + kind (WorksheetColumnKind): `member`: the parent or a member consolidated in full. `combined`: the members' sum, + nothing eliminated. `adjustments`: the consolidation entity's entries. `consolidated`: combined plus the + adjustments. `not_summed`: an equity-method or cost member, shown beside the sum and never in it. + in_sum (bool): Whether the column is part of the consolidated total. + entity_id (None | str | Unset): The entity the column reads; null on a total. + consolidation_method (None | str | Unset): The entity's `consolidation_method`; null on a total. + ownership_pct (float | None | Unset): The parent's share of the entity, where it has one. + """ + + key: str + label: str + kind: WorksheetColumnKind + in_sum: bool + entity_id: None | str | Unset = UNSET + consolidation_method: None | str | Unset = UNSET + ownership_pct: float | None | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + key = self.key + + label = self.label + + kind = self.kind.value + + in_sum = self.in_sum + + entity_id: None | str | Unset + if isinstance(self.entity_id, Unset): + entity_id = UNSET + else: + entity_id = self.entity_id + + consolidation_method: None | str | Unset + if isinstance(self.consolidation_method, Unset): + consolidation_method = UNSET + else: + consolidation_method = self.consolidation_method + + ownership_pct: float | None | Unset + if isinstance(self.ownership_pct, Unset): + ownership_pct = UNSET + else: + ownership_pct = self.ownership_pct + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "key": key, + "label": label, + "kind": kind, + "in_sum": in_sum, + } + ) + if entity_id is not UNSET: + field_dict["entity_id"] = entity_id + if consolidation_method is not UNSET: + field_dict["consolidation_method"] = consolidation_method + if ownership_pct is not UNSET: + field_dict["ownership_pct"] = ownership_pct + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + key = d.pop("key") + + label = d.pop("label") + + kind = WorksheetColumnKind(d.pop("kind")) + + in_sum = d.pop("in_sum") + + def _parse_entity_id(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + entity_id = _parse_entity_id(d.pop("entity_id", UNSET)) + + def _parse_consolidation_method(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + consolidation_method = _parse_consolidation_method( + d.pop("consolidation_method", UNSET) + ) + + def _parse_ownership_pct(data: object) -> float | None | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(float | None | Unset, data) + + ownership_pct = _parse_ownership_pct(d.pop("ownership_pct", UNSET)) + + worksheet_column = cls( + key=key, + label=label, + kind=kind, + in_sum=in_sum, + entity_id=entity_id, + consolidation_method=consolidation_method, + ownership_pct=ownership_pct, + ) + + worksheet_column.additional_properties = d + return worksheet_column + + @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/worksheet_column_kind.py b/robosystems_client/models/worksheet_column_kind.py new file mode 100644 index 0000000..dd67f0b --- /dev/null +++ b/robosystems_client/models/worksheet_column_kind.py @@ -0,0 +1,12 @@ +from enum import Enum + + +class WorksheetColumnKind(str, Enum): + ADJUSTMENTS = "adjustments" + COMBINED = "combined" + CONSOLIDATED = "consolidated" + MEMBER = "member" + NOT_SUMMED = "not_summed" + + def __str__(self) -> str: + return str(self.value) diff --git a/robosystems_client/models/worksheet_row.py b/robosystems_client/models/worksheet_row.py new file mode 100644 index 0000000..734e928 --- /dev/null +++ b/robosystems_client/models/worksheet_row.py @@ -0,0 +1,154 @@ +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="WorksheetRow") + + +@_attrs_define +class WorksheetRow: + """One line of the consolidation worksheet, a value per column. + + Attributes: + qname (str): The rs-gaap concept of the line. + name (str): The line's label. + values (list[float | None]): One value per column, aligned with `columns`. + trait (None | str | Unset): The concept's classification. + depth (int | Unset): Indentation in the statement. Default: 0. + is_subtotal (bool | Unset): Whether the line is a total. Default: False. + period_role (None | str | Unset): On the equity roll-forward, `start` for the beginning-balance row and `end` + for the ending-balance row. Null elsewhere. + """ + + qname: str + name: str + values: list[float | None] + trait: None | str | Unset = UNSET + depth: int | Unset = 0 + is_subtotal: bool | Unset = False + period_role: None | str | Unset = UNSET + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + qname = self.qname + + name = self.name + + values = [] + for values_item_data in self.values: + values_item: float | None + values_item = values_item_data + values.append(values_item) + + trait: None | str | Unset + if isinstance(self.trait, Unset): + trait = UNSET + else: + trait = self.trait + + depth = self.depth + + is_subtotal = self.is_subtotal + + period_role: None | str | Unset + if isinstance(self.period_role, Unset): + period_role = UNSET + else: + period_role = self.period_role + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "qname": qname, + "name": name, + "values": values, + } + ) + if trait is not UNSET: + field_dict["trait"] = trait + if depth is not UNSET: + field_dict["depth"] = depth + if is_subtotal is not UNSET: + field_dict["is_subtotal"] = is_subtotal + if period_role is not UNSET: + field_dict["period_role"] = period_role + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + qname = d.pop("qname") + + name = d.pop("name") + + values = [] + _values = d.pop("values") + for values_item_data in _values: + + def _parse_values_item(data: object) -> float | None: + if data is None: + return data + return cast(float | None, data) + + values_item = _parse_values_item(values_item_data) + + values.append(values_item) + + def _parse_trait(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + trait = _parse_trait(d.pop("trait", UNSET)) + + depth = d.pop("depth", UNSET) + + is_subtotal = d.pop("is_subtotal", UNSET) + + def _parse_period_role(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + period_role = _parse_period_role(d.pop("period_role", UNSET)) + + worksheet_row = cls( + qname=qname, + name=name, + values=values, + trait=trait, + depth=depth, + is_subtotal=is_subtotal, + period_role=period_role, + ) + + worksheet_row.additional_properties = d + return worksheet_row + + @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/tests/test_ledger_client.py b/tests/test_ledger_client.py index 5becea9..d49964f 100644 --- a/tests/test_ledger_client.py +++ b/tests/test_ledger_client.py @@ -2470,6 +2470,40 @@ def test_live_financial_statement_with_period_window( body = mock_op.call_args.kwargs["body"] assert str(body.statement_type) == "income_statement" + @patch("robosystems_client.clients.ledger_client.op_consolidation_worksheet") + def test_consolidation_worksheet(self, mock_op, mock_config, graph_id): + envelope = _envelope( + "consolidation-worksheet", + { + "graph_id": graph_id, + "entity_id": "entity_parent", + "statement_type": "balance_sheet", + "period": {"start": "2026-08-01", "end": "2026-08-31", "label": "Current"}, + "columns": [ + { + "key": "consolidated", + "label": "Consolidated", + "kind": "consolidated", + "in_sum": True, + } + ], + "rows": [], + }, + ) + mock_op.return_value = _mock_response(envelope) + client = LedgerClient(mock_config) + result = client.consolidation_worksheet( + graph_id, + { + "statement_type": "balance_sheet", + "period_start": "2026-08-01", + "period_end": "2026-08-31", + }, + ) + assert result["columns"][0]["kind"] == "consolidated" + body = mock_op.call_args.kwargs["body"] + assert str(body.statement_type) == "balance_sheet" + @patch("robosystems_client.clients.ledger_client.op_live_financial_statement") def test_live_financial_statement_with_fiscal_year( self, mock_op, mock_config, graph_id