feat: Support transient identities and traits (#93)

* feat: Support transient identities and traits
- Support transient identities and traits
- Bump requests
- Bump mypy
- Remove linting from CI in favour of pre-commit.ci

Author: khvn26

.github/workflows/pytest.yml

@@ -1,4 +1,4 @@
-name: Linting and Tests
+name: Run Tests
 
 on:
   pull_request:
@@ -9,7 +9,6 @@ on:
 jobs:
   test:
     runs-on: ubuntu-latest
-    name: Linting and Tests
 
     strategy:
       max-parallel: 4
@@ -33,14 +32,5 @@ jobs:
           pip install poetry
           poetry install --with dev
 
-      - name: Check Formatting
-        run: |
-          poetry run black --check .
-          poetry run flake8 .
-          poetry run isort --check .
-
-      - name: Check Typing
-        run: poetry run mypy --strict .
-
       - name: Run Tests
         run: poetry run pytest

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.5.1
+    rev: v1.10.1
     hooks:
       - id: mypy
         args: [--strict]
@@ -14,7 +14,6 @@ repos:
     rev: 24.3.0
     hooks:
       - id: black
-        language_version: python3
   - repo: https://github.com/pycqa/flake8
     rev: 6.1.0
     hooks:

README.md

@@ -14,8 +14,8 @@ For full documentation visit
 
 ## Contributing
 
-Please read [CONTRIBUTING.md](CONTRIBUTING.md) for details on our code
-of conduct, and the process for submitting pull requests
+Please read [CONTRIBUTING.md](CONTRIBUTING.md) for details on our code of conduct, and the process for submitting pull
+requests
 
 ## Getting Help
 

flagsmith/flagsmith.py

@@ -19,23 +19,14 @@
 from flagsmith.offline_handlers import BaseOfflineHandler
 from flagsmith.polling_manager import EnvironmentDataPollingManager
 from flagsmith.streaming_manager import EventStreamManager, StreamEvent
-from flagsmith.utils.identities import Identity, generate_identities_data
+from flagsmith.types import JsonType, TraitConfig, TraitMapping
+from flagsmith.utils.identities import generate_identity_data
 
 logger = logging.getLogger(__name__)
 
 DEFAULT_API_URL = "https://edge.api.flagsmith.com/api/v1/"
 DEFAULT_REALTIME_API_URL = "https://realtime.flagsmith.com/"
 
-JsonType = typing.Union[
-    None,
-    int,
-    str,
-    bool,
-    typing.List["JsonType"],
-    typing.List[typing.Mapping[str, "JsonType"]],
-    typing.Dict[str, "JsonType"],
-]
-
 
 class Flagsmith:
     """A Flagsmith client.
@@ -237,7 +228,9 @@ def get_environment_flags(self) -> Flags:
     def get_identity_flags(
         self,
         identifier: str,
-        traits: typing.Optional[typing.Mapping[str, TraitValue]] = None,
+        traits: typing.Optional[TraitMapping] = None,
+        *,
+        transient: bool = False,
     ) -> Flags:
         """
         Get all the flags for the current environment for a given identity. Will also
@@ -247,13 +240,20 @@ def get_identity_flags(
         :param identifier: a unique identifier for the identity in the current
             environment, e.g. email address, username, uuid
         :param traits: a dictionary of traits to add / update on the identity in
-            Flagsmith, e.g. {"num_orders": 10}
+            Flagsmith, e.g. `{"num_orders": 10}`. Envelope traits you don't want persisted
+            in a dictionary with `"transient"` and `"value"` keys, e.g.
+            `{"num_orders": 10, "color": {"value": "pink", "transient": True}}`.
+        :param transient: if `True`, the identity won't get persisted
         :return: Flags object holding all the flags for the given identity.
         """
         traits = traits or {}
         if (self.offline_mode or self.enable_local_evaluation) and self._environment:
             return self._get_identity_flags_from_document(identifier, traits)
-        return self._get_identity_flags_from_api(identifier, traits)
+        return self._get_identity_flags_from_api(
+            identifier,
+            traits,
+            transient=transient,
+        )
 
     def get_identity_segments(
         self,
@@ -306,7 +306,7 @@ def _get_environment_flags_from_document(self) -> Flags:
         )
 
     def _get_identity_flags_from_document(
-        self, identifier: str, traits: typing.Mapping[str, TraitValue]
+        self, identifier: str, traits: TraitMapping
     ) -> Flags:
         identity_model = self._get_identity_model(identifier, **traits)
         if self._environment is None:
@@ -339,13 +339,23 @@ def _get_environment_flags_from_api(self) -> Flags:
             raise
 
     def _get_identity_flags_from_api(
-        self, identifier: str, traits: typing.Mapping[str, typing.Any]
+        self,
+        identifier: str,
+        traits: TraitMapping,
+        *,
+        transient: bool = False,
     ) -> Flags:
+        request_body = generate_identity_data(
+            identifier,
+            traits,
+            transient=transient,
+        )
         try:
-            data = generate_identities_data(identifier, traits)
             json_response: typing.Dict[str, typing.List[typing.Dict[str, JsonType]]] = (
                 self._get_json_response(
-                    url=self.identities_url, method="POST", body=data
+                    url=self.identities_url,
+                    method="POST",
+                    body=request_body,
                 )
             )
             return Flags.from_api_flags(
@@ -364,9 +374,7 @@ def _get_json_response(
         self,
         url: str,
         method: str,
-        body: typing.Optional[
-            typing.Union[Identity, typing.Dict[str, JsonType]]
-        ] = None,
+        body: typing.Optional[JsonType] = None,
     ) -> typing.Any:
         try:
             request_method = getattr(self.session, method.lower())
@@ -387,15 +395,18 @@ def _get_json_response(
     def _get_identity_model(
         self,
         identifier: str,
-        **traits: TraitValue,
+        **traits: typing.Union[TraitValue, TraitConfig],
     ) -> IdentityModel:
         if not self._environment:
             raise FlagsmithClientError(
                 "Unable to build identity model when no local environment present."
             )
 
         trait_models = [
-            TraitModel(trait_key=key, trait_value=value)
+            TraitModel(
+                trait_key=key,
+                trait_value=value["value"] if isinstance(value, dict) else value,
+            )
             for key, value in traits.items()
         ]
 

flagsmith/streaming_manager.py

@@ -22,7 +22,7 @@ def __init__(
         stream_url: str,
         on_event: Callable[[StreamEvent], None],
         request_timeout_seconds: Optional[int] = None,
-        **kwargs: typing.Any
+        **kwargs: typing.Any,
     ) -> None:
         super().__init__(*args, **kwargs)
         self._stop_event = threading.Event()

flagsmith/types.py

@@ -0,0 +1,25 @@
+import typing
+
+from flag_engine.identities.traits.types import TraitValue
+from typing_extensions import TypeAlias
+
+_JsonScalarType: TypeAlias = typing.Union[
+    int,
+    str,
+    float,
+    bool,
+    None,
+]
+JsonType: TypeAlias = typing.Union[
+    _JsonScalarType,
+    typing.Dict[str, "JsonType"],
+    typing.List["JsonType"],
+]
+
+
+class TraitConfig(typing.TypedDict):
+    value: TraitValue
+    transient: bool
+
+
+TraitMapping: TypeAlias = typing.Mapping[str, typing.Union[TraitValue, TraitConfig]]

flagsmith/utils/identities.py

@@ -1,21 +1,26 @@
 import typing
 
-from flag_engine.identities.traits.types import TraitValue
+from flagsmith.types import JsonType, TraitMapping
 
-Identity = typing.TypedDict(
-    "Identity",
-    {"identifier": str, "traits": typing.List[typing.Mapping[str, TraitValue]]},
-)
 
-
-def generate_identities_data(
-    identifier: str, traits: typing.Optional[typing.Mapping[str, TraitValue]] = None
-) -> Identity:
-    return {
-        "identifier": identifier,
-        "traits": (
-            [{"trait_key": k, "trait_value": v} for k, v in traits.items()]
-            if traits
-            else []
-        ),
-    }
+def generate_identity_data(
+    identifier: str,
+    traits: TraitMapping,
+    *,
+    transient: bool,
+) -> JsonType:
+    identity_data: typing.Dict[str, JsonType] = {"identifier": identifier}
+    traits_data: typing.List[JsonType] = []
+    for trait_key, trait_value in traits.items():
+        trait_data: typing.Dict[str, JsonType] = {"trait_key": trait_key}
+        if isinstance(trait_value, dict):
+            trait_data["trait_value"] = trait_value["value"]
+            if trait_value.get("transient"):
+                trait_data["transient"] = True
+        else:
+            trait_data["trait_value"] = trait_value
+        traits_data.append(trait_data)
+    identity_data["traits"] = traits_data
+    if transient:
+        identity_data["transient"] = True
+    return identity_data