feat(python): Add ListFlags api (#127)

flipt-python/flipt/async_client.py

@@ -2,6 +2,7 @@
 
 from .authentication import AuthenticationStrategy
 from .evaluation import AsyncEvaluation
+from .flags import AsyncFlag
 
 
 class AsyncFliptClient:
@@ -14,6 +15,7 @@ def __init__(
         self.httpx_client = httpx.AsyncClient(timeout=timeout)
 
         self.evaluation = AsyncEvaluation(url, authentication, self.httpx_client)
+        self.flag = AsyncFlag(url, authentication, self.httpx_client)
 
     async def close(self) -> None:
         await self.httpx_client.aclose()

flipt-python/flipt/evaluation/models.py

@@ -1,7 +1,8 @@
 from enum import StrEnum
 
-from pydantic import AliasGenerator, BaseModel, ConfigDict, Field
-from pydantic.alias_generators import to_camel
+from pydantic import Field
+
+from flipt.models import CamelAliasModel
 
 
 class EvaluationResponseType(StrEnum):
@@ -22,13 +23,6 @@ class ErrorEvaluationReason(StrEnum):
     NOT_FOUND_ERROR_EVALUATION_REASON = "NOT_FOUND_ERROR_EVALUATION_REASON"
 
 
-class CamelAliasModel(BaseModel):
-    model_config = ConfigDict(
-        alias_generator=AliasGenerator(alias=to_camel),
-        populate_by_name=True,
-    )
-
-
 class EvaluationRequest(CamelAliasModel):
     namespace_key: str = Field(default="default")
     flag_key: str

flipt-python/flipt/flags/__init__.py

@@ -0,0 +1,15 @@
+from .async_flag_client import AsyncFlag
+from .models import (
+    Flag,
+    FlagType,
+    ListFlagsResponse,
+)
+from .sync_flag_client import SyncFlag
+
+__all__ = [
+    "AsyncFlag",
+    "SyncFlag",
+    "ListFlagsResponse",
+    "Flag",
+    "FlagType",
+]

flipt-python/flipt/flags/async_flag_client.py

@@ -0,0 +1,45 @@
+from http import HTTPStatus
+
+import httpx
+
+from flipt.models import ListParameters
+
+from ..authentication import AuthenticationStrategy
+from ..exceptions import FliptApiError
+from .models import (
+    ListFlagsResponse,
+)
+
+
+class AsyncFlag:
+    def __init__(
+        self,
+        url: str,
+        authentication: AuthenticationStrategy | None = None,
+        httpx_client: httpx.AsyncClient | None = None,
+    ):
+        self.url = url
+        self.headers: dict[str, str] = {}
+
+        self._client = httpx_client or httpx.AsyncClient()
+
+        if authentication:
+            authentication.authenticate(self.headers)
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+    def _raise_on_error(self, response: httpx.Response) -> None:
+        if response.status_code != 200:
+            body = response.json()
+            message = body.get("message", HTTPStatus(response.status_code).description)
+            raise FliptApiError(message, response.status_code)
+
+    async def list_flags(self, *, namespace_key: str, params: ListParameters | None = None) -> ListFlagsResponse:
+        response = await self._client.get(
+            f"{self.url}/api/v1/namespaces/{namespace_key}/flags",
+            params=params.model_dump_json(exclude_none=True) if params else {},
+            headers=self.headers,
+        )
+        self._raise_on_error(response)
+        return ListFlagsResponse.model_validate_json(response.text)

flipt-python/flipt/flags/models.py

@@ -0,0 +1,26 @@
+from datetime import datetime
+from enum import Enum
+from typing import Any
+
+from flipt.models import CamelAliasModel, PaginatedResponse
+
+
+class FlagType(str, Enum):
+    variant = "VARIANT_FLAG_TYPE"
+    boolean = "BOOLEAN_FLAG_TYPE"
+
+
+class Flag(CamelAliasModel):
+    created_at: datetime
+    description: str
+    enabled: bool
+    key: str
+    name: str
+    namespacekey: str | None = None
+    type: FlagType
+    updatedAt: datetime
+    variants: list[Any]
+
+
+class ListFlagsResponse(CamelAliasModel, PaginatedResponse):
+    flags: list[Flag]

flipt-python/flipt/flags/sync_flag_client.py

@@ -0,0 +1,45 @@
+from http import HTTPStatus
+
+import httpx
+
+from flipt.models import ListParameters
+
+from ..authentication import AuthenticationStrategy
+from ..exceptions import FliptApiError
+from .models import (
+    ListFlagsResponse,
+)
+
+
+class SyncFlag:
+    def __init__(
+        self,
+        url: str,
+        authentication: AuthenticationStrategy | None = None,
+        httpx_client: httpx.Client | None = None,
+    ):
+        self.url = url
+        self.headers: dict[str, str] = {}
+
+        self._client = httpx_client or httpx.Client()
+
+        if authentication:
+            authentication.authenticate(self.headers)
+
+    def close(self) -> None:
+        self._client.close()
+
+    def _raise_on_error(self, response: httpx.Response) -> None:
+        if response.status_code != 200:
+            body = response.json()
+            message = body.get("message", HTTPStatus(response.status_code).description)
+            raise FliptApiError(message, response.status_code)
+
+    def list_flags(self, *, namespace_key: str, params: ListParameters | None = None) -> ListFlagsResponse:
+        response = self._client.get(
+            f"{self.url}/api/v1/namespaces/{namespace_key}/flags",
+            params=params.model_dump_json(exclude_none=True) if params else {},
+            headers=self.headers,
+        )
+        self._raise_on_error(response)
+        return ListFlagsResponse.model_validate_json(response.text)

flipt-python/flipt/models.py

@@ -0,0 +1,21 @@
+from pydantic import AliasGenerator, BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+
+
+class CamelAliasModel(BaseModel):
+    model_config = ConfigDict(
+        alias_generator=AliasGenerator(alias=to_camel),
+        populate_by_name=True,
+    )
+
+
+class ListParameters(BaseModel):
+    limit: int | None = None
+    offset: int | None = None
+    pageToken: str | None = None
+    reference: str | None = None
+
+
+class PaginatedResponse(BaseModel):
+    nextPageToken: str
+    totalCount: int

flipt-python/flipt/sync_client.py

@@ -2,6 +2,7 @@
 
 from .authentication import AuthenticationStrategy
 from .evaluation import Evaluation
+from .flags import SyncFlag
 
 
 class FliptClient:
@@ -14,6 +15,7 @@ def __init__(
         self.httpx_client = httpx.Client(timeout=timeout)
 
         self.evaluation = Evaluation(url, authentication, self.httpx_client)
+        self.flag = SyncFlag(url, authentication, self.httpx_client)
 
     def close(self) -> None:
         self.httpx_client.close()

flipt-python/tests/conftest.py

@@ -6,15 +6,15 @@
 from flipt.authentication import ClientTokenAuthentication
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
 def flipt_url() -> str:
     flipt_url = os.environ.get("FLIPT_URL")
     if flipt_url is None:
         raise Exception("FLIPT_URL not set")
     return flipt_url
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
 def flipt_auth_token() -> str:
     auth_token = os.environ.get("FLIPT_AUTH_TOKEN")
     if auth_token is None:
@@ -23,7 +23,7 @@ def flipt_auth_token() -> str:
     return auth_token
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
 def sync_flipt_client(flipt_url, flipt_auth_token):
     return FliptClient(url=flipt_url, authentication=ClientTokenAuthentication(flipt_auth_token))
 

flipt-python/tests/flag/conftest.py

@@ -0,0 +1,13 @@
+from http import HTTPStatus
+
+import pytest
+
+
+@pytest.fixture(params=[{}, {'message': 'some error'}])
+def _mock_list_flags_response_error(httpx_mock, flipt_url, request):
+    httpx_mock.add_response(
+        method="GET",
+        url=f'{flipt_url}/api/v1/namespaces/default/flags',
+        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+        json=request.param,
+    )

flipt-python/tests/flag/test_async_client.py

@@ -0,0 +1,15 @@
+import pytest
+
+from flipt.async_client import AsyncFliptClient
+from flipt.exceptions import FliptApiError
+
+
+class TestListFlags:
+    async def test_success(self, async_flipt_client: AsyncFliptClient):
+        list_response = await async_flipt_client.flag.list_flags(namespace_key="default")
+        assert len(list_response.flags) == 2
+
+    @pytest.mark.usefixtures("_mock_list_flags_response_error")
+    async def test_list_error(self, async_flipt_client):
+        with pytest.raises(FliptApiError):
+            await async_flipt_client.flag.list_flags(namespace_key="default")

flipt-python/tests/flag/test_sync_client.py

@@ -0,0 +1,14 @@
+import pytest
+
+from flipt.exceptions import FliptApiError
+
+
+class TestListFlags:
+    def test_success(self, sync_flipt_client):
+        list_response = sync_flipt_client.flag.list_flags(namespace_key="default")
+        assert len(list_response.flags) == 2
+
+    @pytest.mark.usefixtures("_mock_list_flags_response_error")
+    def test_list_error(self, sync_flipt_client):
+        with pytest.raises(FliptApiError):
+            sync_flipt_client.flag.list_flags(namespace_key="default")