Skip to content

Improve typing #121

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pbrassel merged 1 commit into from
Jul 21, 2026
Merged

Improve typing #121

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/types_api.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,10 @@ Types
.. autodata:: flame_hub._base_client.ResourceT
:annotation: = typing.TypeVar

.. autoclass:: flame_hub._base_client.ResourceListResult
:members:
:undoc-members:

Comment thread
pbrassel marked this conversation as resolved.
.. autoclass:: flame_hub.types.FieldParams

.. autoclass:: flame_hub.types.IncludeParams
Expand Down
61 changes: 31 additions & 30 deletions flame_hub/_auth_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
get_includable_names,
UNSET,
UNSET_T,
ResourceListResult,
)
from flame_hub._defaults import DEFAULT_AUTH_BASE_URL
from flame_hub._auth_flows import ClientAuth, PasswordAuth
Expand Down Expand Up @@ -229,18 +230,18 @@ class AuthClient(BaseClient):
def __init__(
self,
base_url=DEFAULT_AUTH_BASE_URL,
auth: ClientAuth | PasswordAuth = None,
auth: ClientAuth | PasswordAuth | None = None,
**kwargs: te.Unpack[ClientKwargs],
):
super().__init__(base_url, auth, **kwargs)

def get_realms(self, **params: te.Unpack[GetKwargs]) -> list[Realm]:
def get_realms(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[Realm]:
return self._get_all_resources(Realm, "realms", **params)

def find_realms(self, **params: te.Unpack[FindAllKwargs]) -> list[Realm]:
def find_realms(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[Realm]:
return self._find_all_resources(Realm, "realms", **params)

def create_realm(self, name: str, display_name: str = None, description: str = None) -> Realm:
def create_realm(self, name: str, display_name: str | None = None, description: str | None = None) -> Realm:
return self._create_resource(
Realm,
CreateRealm(
Expand Down Expand Up @@ -278,9 +279,9 @@ def update_realm(
def create_permission(
self,
name: str,
display_name: str = None,
description: str = None,
realm_id: Realm | uuid.UUID | str = None,
display_name: str | None = None,
description: str | None = None,
realm_id: Realm | uuid.UUID | str | None = None,
) -> Permission:
return self._create_resource(
Permission,
Expand Down Expand Up @@ -319,13 +320,13 @@ def update_permission(
permission_id,
)

def get_permissions(self, **params: te.Unpack[GetKwargs]) -> list[Permission]:
def get_permissions(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[Permission]:
return self._get_all_resources(Permission, "permissions", include=get_includable_names(Permission), **params)

def find_permissions(self, **params: te.Unpack[FindAllKwargs]) -> list[Permission]:
def find_permissions(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[Permission]:
return self._find_all_resources(Permission, "permissions", include=get_includable_names(Permission), **params)

def create_role(self, name: str, display_name: str = None, description: str = None) -> Role:
def create_role(self, name: str, display_name: str | None = None, description: str | None = None) -> Role:
return self._create_resource(
Role,
CreateRole(name=name, display_name=display_name, description=description),
Expand All @@ -352,10 +353,10 @@ def update_role(
role_id,
)

def get_roles(self, **params: te.Unpack[GetKwargs]) -> list[Role]:
def get_roles(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[Role]:
return self._get_all_resources(Role, "roles", include=get_includable_names(Role), **params)

def find_roles(self, **params: te.Unpack[FindAllKwargs]) -> list[Role]:
def find_roles(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[Role]:
return self._find_all_resources(Role, "roles", include=get_includable_names(Role), **params)

def create_role_permission(
Expand All @@ -381,15 +382,15 @@ def get_role_permission(
def delete_role_permission(self, role_permission_id: RolePermission | uuid.UUID | str):
self._delete_resource("role-permissions", role_permission_id)

def get_role_permissions(self, **params: te.Unpack[GetKwargs]) -> list[RolePermission]:
def get_role_permissions(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[RolePermission]:
return self._get_all_resources(
RolePermission,
"role-permissions",
include=get_includable_names(RolePermission),
**params,
)

def find_role_permissions(self, **params: te.Unpack[FindAllKwargs]) -> list[RolePermission]:
def find_role_permissions(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[RolePermission]:
return self._find_all_resources(
RolePermission,
"role-permissions",
Expand All @@ -401,11 +402,11 @@ def create_user(
self,
name: str,
email: str,
display_name: str = None,
display_name: str | None = None,
active: bool = True,
name_locked: bool = False,
first_name: str = None,
last_name: str = None,
first_name: str | None = None,
last_name: str | None = None,
) -> User:
return self._create_resource(
User,
Expand Down Expand Up @@ -453,10 +454,10 @@ def update_user(
user_id,
)

def get_users(self, **params: te.Unpack[GetKwargs]) -> list[User]:
def get_users(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[User]:
return self._get_all_resources(User, "users", include=get_includable_names(User), **params)

def find_users(self, **params: te.Unpack[FindAllKwargs]) -> list[User]:
def find_users(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[User]:
return self._find_all_resources(User, "users", include=get_includable_names(User), **params)

def create_user_permission(
Expand Down Expand Up @@ -484,15 +485,15 @@ def get_user_permission(
def delete_user_permission(self, user_permission_id: UserPermission | uuid.UUID | str):
self._delete_resource("user-permissions", user_permission_id)

def get_user_permissions(self, **params: te.Unpack[GetKwargs]) -> list[UserPermission]:
def get_user_permissions(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[UserPermission]:
return self._get_all_resources(
UserPermission,
"user-permissions",
include=get_includable_names(UserPermission),
**params,
)

def find_user_permissions(self, **params: te.Unpack[FindAllKwargs]) -> list[UserPermission]:
def find_user_permissions(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[UserPermission]:
return self._find_all_resources(
UserPermission,
"user-permissions",
Expand All @@ -517,24 +518,24 @@ def get_user_role(
def delete_user_role(self, user_role_id: UserRole | uuid.UUID | str):
self._delete_resource("user-roles", user_role_id)

def get_user_roles(self, **params: te.Unpack[GetKwargs]) -> list[UserRole]:
def get_user_roles(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[UserRole]:
return self._get_all_resources(UserRole, "user-roles", include=get_includable_names(UserRole), **params)

def find_user_roles(self, **params: te.Unpack[FindAllKwargs]) -> list[UserRole]:
def find_user_roles(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[UserRole]:
return self._find_all_resources(UserRole, "user-roles", include=get_includable_names(UserRole), **params)

def create_client(
self,
name: str,
realm_id: Realm | str | uuid.UUID,
secret: str = None,
display_name: str = None,
description: str = None,
redirect_uri: str = None,
secret: str | None = None,
display_name: str | None = None,
description: str | None = None,
redirect_uri: str | None = None,
active: bool = True,
is_confidential: bool = True,
secret_hashed: bool = False,
grant_types: str = None,
grant_types: str | None = None,
) -> Client:
return self._create_resource(
Client,
Expand All @@ -559,10 +560,10 @@ def delete_client(self, client_id: Client | uuid.UUID | str):
def get_client(self, client_id: Client | uuid.UUID | str, **params: te.Unpack[GetKwargs]) -> Client | None:
return self._get_single_resource(Client, "clients", client_id, include=get_includable_names(Client), **params)

def get_clients(self, **params: te.Unpack[GetKwargs]) -> list[Client]:
def get_clients(self, **params: te.Unpack[GetKwargs]) -> ResourceListResult[Client]:
return self._get_all_resources(Client, "clients", include=get_includable_names(Client), **params)

def find_clients(self, **params: te.Unpack[FindAllKwargs]) -> list[Client]:
def find_clients(self, **params: te.Unpack[FindAllKwargs]) -> ResourceListResult[Client]:
return self._find_all_resources(Client, "clients", include=get_includable_names(Client), **params)

def update_client(
Expand Down
21 changes: 14 additions & 7 deletions flame_hub/_base_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,12 @@ class UuidModel(t.Protocol[ResourceT]):

id: uuid.UUID

def __str__(self) -> str:
return str(self.id)


# union which encompasses all types where a UUID can be extracted from
UuidIdentifiable = UuidModel | uuid.UUID | str
UuidIdentifiable: t.TypeAlias = UuidModel | uuid.UUID | str


def obtain_uuid_from(uuid_identifiable: UuidIdentifiable) -> uuid.UUID:
Expand Down Expand Up @@ -115,6 +118,10 @@ class ResourceList(BaseModel, t.Generic[ResourceT]):
"""Attribute which holds meta information about the result and the requested resource type."""


# Generic type alias for all get and find methods.
ResourceListResult: t.TypeAlias = list[ResourceT] | tuple[list[ResourceT], ResourceListMeta]


class SortParams(te.TypedDict, total=False):
"""Dict shape for specifying parameters for sorted queries.

Expand Down Expand Up @@ -285,7 +292,7 @@ def build_page_params(page_params: PageParams | None = None, default_page_params
page_params: PageParams = {}

# overwrite default values with user-defined ones
page_params = default_page_params | page_params
page_params = {**default_page_params, **page_params}

return {f"page[{k}]": v for k, v in page_params.items()}

Expand Down Expand Up @@ -332,7 +339,7 @@ def build_sort_params(sort_params: SortParams | None = None) -> dict:
# property gets a "-" prepended if sorting in descending order
param_sort_prefix = "-" if param_sort_order == "descending" else ""
# construct the actual query params
query_params["sort"] = param_sort_prefix + param_sort_by
query_params["sort"] = f"{param_sort_prefix}{param_sort_by}"

return query_params

Expand Down Expand Up @@ -417,7 +424,7 @@ def _get_all_resources(
include: IncludeParams | None = None,
expected_code: int = httpx.codes.OK.value,
**params: te.Unpack[GetKwargs],
) -> list[ResourceT] | tuple[list[ResourceT], ResourceListMeta]:
) -> ResourceListResult[ResourceT]:
"""Retrieve all resources of a certain type at the specified path from the FLAME Hub.

This method passes its arguments through to :py:meth:`_find_all_resources`. Check the documentation of that
Expand All @@ -441,7 +448,7 @@ def _find_all_resources(
include: IncludeParams | None = None,
expected_code: int = httpx.codes.OK.value,
**params: te.Unpack[FindAllKwargs],
) -> list[ResourceT] | tuple[list[ResourceT], ResourceListMeta]:
) -> ResourceListResult[ResourceT]:
"""Find all resources at the specified path on the FLAME Hub that match certain criteria.

This method accesses the endpoint ``*path`` and returns all resources of type ``resource_type`` that match
Expand Down Expand Up @@ -492,7 +499,7 @@ def _find_all_resources(
filter_params = params.get("filter", None)
sort_params = params.get("sort", None)
field_params = params.get("fields", None)
meta = params.get("meta", False)
meta_flag = params.get("meta", False)

request_params = (
build_page_params(page_params)
Expand All @@ -509,7 +516,7 @@ def _find_all_resources(

resource_list = ResourceList[resource_type](**r.json())

if meta:
if meta_flag:
return resource_list.data, resource_list.meta
else:
return resource_list.data
Expand Down
Loading