From 48114b434ac8862aa4e3623c5513bfcb30c607e0 Mon Sep 17 00:00:00 2001 From: Richard Dzurus Date: Fri, 23 May 2025 00:40:24 +0200 Subject: [PATCH] polyCustom - prevent rewrites of executionId --- polyapi/__init__.py | 83 +++++++++++++++++++++++++++++++++++++++++---- pyproject.toml | 2 +- 2 files changed, 77 insertions(+), 8 deletions(-) diff --git a/polyapi/__init__.py b/polyapi/__init__.py index 2a30c36..7935843 100644 --- a/polyapi/__init__.py +++ b/polyapi/__init__.py @@ -1,7 +1,8 @@ import os import sys import truststore -from typing import Dict, Any +from typing import Any, Optional, overload, Literal +from typing_extensions import TypedDict truststore.inject_into_ssl() from .cli import CLI_COMMANDS @@ -15,9 +16,77 @@ sys.exit(1) -polyCustom: Dict[str, Any] = { - "executionId": None, - "executionApiKey": None, - "responseStatusCode": 200, - "responseContentType": None, -} \ No newline at end of file +class PolyCustomDict(TypedDict, total=False): + """Type definition for polyCustom dictionary.""" + executionId: Optional[str] # Read-only + executionApiKey: Optional[str] + responseStatusCode: int + responseContentType: Optional[str] + + +class _PolyCustom: + def __init__(self): + self._internal_store = { + "executionId": None, + "executionApiKey": None, + "responseStatusCode": 200, + "responseContentType": None, + } + self._execution_id_locked = False + + def set_once(self, key: str, value: Any) -> None: + if key == "executionId" and self._execution_id_locked: + # Silently ignore attempts to overwrite locked executionId + return + self._internal_store[key] = value + if key == "executionId": + # Lock executionId after setting it + self.lock_execution_id() + + def get(self, key: str, default: Any = None) -> Any: + return self._internal_store.get(key, default) + + def lock_execution_id(self) -> None: + self._execution_id_locked = True + + def unlock_execution_id(self) -> None: + self._execution_id_locked = False + + @overload + def __getitem__(self, key: Literal["executionId"]) -> Optional[str]: ... + + @overload + def __getitem__(self, key: Literal["executionApiKey"]) -> Optional[str]: ... + + @overload + def __getitem__(self, key: Literal["responseStatusCode"]) -> int: ... + + @overload + def __getitem__(self, key: Literal["responseContentType"]) -> Optional[str]: ... + + def __getitem__(self, key: str) -> Any: + return self.get(key) + + @overload + def __setitem__(self, key: Literal["executionApiKey"], value: Optional[str]) -> None: ... + + @overload + def __setitem__(self, key: Literal["responseStatusCode"], value: int) -> None: ... + + @overload + def __setitem__(self, key: Literal["responseContentType"], value: Optional[str]) -> None: ... + + def __setitem__(self, key: str, value: Any) -> None: + self.set_once(key, value) + + def __repr__(self) -> str: + return f"PolyCustom({self._internal_store})" + + def copy(self) -> '_PolyCustom': + new = _PolyCustom() + new._internal_store = self._internal_store.copy() + new._execution_id_locked = self._execution_id_locked + return new + + +polyCustom: PolyCustomDict = _PolyCustom() \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 61f6eec..bb639af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,7 +3,7 @@ requires = ["setuptools>=61.2", "wheel"] [project] name = "polyapi-python" -version = "0.3.7.dev1" +version = "0.3.7.dev2" description = "The Python Client for PolyAPI, the IPaaS by Developers for Developers" authors = [{ name = "Dan Fellin", email = "dan@polyapi.io" }] dependencies = [