Skip to content

Chore: Type hints cleanup #272

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
iLLiCiTiT merged 5 commits into from
Sep 10, 2025
Merged

Chore: Type hints cleanup #272

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
633c3a8
fixes of type hints in utils
iLLiCiTiT Sep 8, 2025
73cb911
unified type hints in server api
iLLiCiTiT Sep 8, 2025
3e6259e
add spaces between variable and default value
iLLiCiTiT Sep 8, 2025
c110ff9
added type hints to as user stact
iLLiCiTiT Sep 8, 2025
cbc774c
Merge branch 'develop' into enhancement/type-hints-cleanup
iLLiCiTiT Sep 8, 2025
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
33 changes: 16 additions & 17 deletions ayon_api/server_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
import uuid
from contextlib import contextmanager
import typing
from typing import Optional, Iterable, Generator, Any
from typing import Optional, Iterable, Generator, Any, Union

import requests

Expand Down Expand Up @@ -87,7 +87,6 @@
)

if typing.TYPE_CHECKING:
from typing import Union
from .typing import (
ServerVersion,
AnyEntityDict,
Expand Down Expand Up @@ -163,23 +162,23 @@ def clear(self):
self._default_user = None

@property
def username(self):
def username(self) -> Optional[str]:
# Use '_user_ids' for boolean check to have ability "unset"
# default user
if self._user_ids:
return self._last_user
return self._default_user

def get_default_username(self):
def get_default_username(self) -> Optional[str]:
return self._default_user

def set_default_username(self, username=None):
def set_default_username(self, username: Optional[str] = None) -> None:
self._default_user = username

default_username = property(get_default_username, set_default_username)

@contextmanager
def as_user(self, username):
def as_user(self, username: Optional[str]) -> Generator[None, None, None]:
self._last_user = username
user_id = uuid.uuid4().hex
self._user_ids.append(user_id)
Expand Down Expand Up @@ -273,7 +272,7 @@ def __init__(
default_settings_variant: Optional[str] = None,
sender_type: Optional[str] = None,
sender: Optional[str] = None,
ssl_verify: Optional["Union[bool, str]"]=None,
ssl_verify: Optional[Union[bool, str]] = None,
cert: Optional[str] = None,
create_session: bool = True,
timeout: Optional[float] = None,
Expand Down Expand Up @@ -685,15 +684,15 @@ def set_default_service_username(self, username: Optional[str] = None):
@contextmanager
def as_username(
self,
username: "Union[str, None]",
username: Optional[str],
ignore_service_error: bool = False,
):
"""Service API will temporarily work as other user.

This method can be used only if service API key is logged in.

Args:
username (Union[str, None]): Username to work as when service.
username (Optional[str]): Username to work as when service.
ignore_service_error (Optional[bool]): Ignore error when service
API key is not used.

Expand Down Expand Up @@ -873,7 +872,7 @@ def get_server_version(self) -> str:
self._server_version = self.get_info()["version"]
return self._server_version

def get_server_version_tuple(self) -> "ServerVersion":
def get_server_version_tuple(self) -> ServerVersion:
"""Get server version as tuple.

Version should match semantic version (https://semver.org/).
Expand All @@ -897,7 +896,7 @@ def get_server_version_tuple(self) -> "ServerVersion":
return self._server_version_tuple

server_version = property(get_server_version)
server_version_tuple: "ServerVersion" = property(
server_version_tuple: ServerVersion = property(
get_server_version_tuple
)

Expand Down Expand Up @@ -1360,7 +1359,7 @@ def _download_file_to_stream(
def download_file_to_stream(
self,
endpoint: str,
stream: "StreamType",
stream: StreamType,
chunk_size: Optional[int] = None,
progress: Optional[TransferProgress] = None,
) -> TransferProgress:
Expand Down Expand Up @@ -1462,7 +1461,7 @@ def download_file(

@staticmethod
def _upload_chunks_iter(
file_stream: "StreamType",
file_stream: StreamType,
progress: TransferProgress,
chunk_size: int,
) -> Generator[bytes, None, None]:
Expand Down Expand Up @@ -1494,7 +1493,7 @@ def _upload_chunks_iter(
def _upload_file(
self,
url: str,
stream: "StreamType",
stream: StreamType,
progress: TransferProgress,
request_type: Optional[RequestType] = None,
chunk_size: Optional[int] = None,
Expand Down Expand Up @@ -1545,7 +1544,7 @@ def _upload_file(
def upload_file_from_stream(
self,
endpoint: str,
stream: "StreamType",
stream: StreamType,
progress: Optional[TransferProgress] = None,
request_type: Optional[RequestType] = None,
**kwargs
Expand Down Expand Up @@ -1842,7 +1841,7 @@ def get_rest_entity_by_id(
project_name: str,
entity_type: str,
entity_id: str,
) -> Optional["AnyEntityDict"]:
) -> Optional[AnyEntityDict]:
"""Get entity using REST on a project by its id.

Args:
Expand Down Expand Up @@ -2029,7 +2028,7 @@ def _prepare_fields(
)
}

def _convert_entity_data(self, entity: "AnyEntityDict"):
def _convert_entity_data(self, entity: AnyEntityDict):
if not entity or "data" not in entity:
return

Expand Down
35 changes: 17 additions & 18 deletions ayon_api/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
import itertools
from urllib.parse import urlparse, urlencode
import typing
from typing import Optional, Dict, Set, Any, Iterable
from typing import Optional, Any, Iterable, Union
from enum import IntEnum

import requests
Expand All @@ -38,7 +38,6 @@


if typing.TYPE_CHECKING:
from typing import Union
from .typing import AnyEntityDict, StreamType

REMOVED_VALUE = object()
Expand Down Expand Up @@ -206,7 +205,7 @@ def get(self, key, default=None):
return default


def fill_own_attribs(entity: "AnyEntityDict") -> None:
def fill_own_attribs(entity: AnyEntityDict) -> None:
"""Fill own attributes.

Prepare data with own attributes. Prepare data based on a list of
Expand Down Expand Up @@ -353,7 +352,7 @@ def is_valid(self) -> bool:


def prepare_query_string(
key_values: Dict[str, Any], skip_none: bool = True
key_values: dict[str, Any], skip_none: bool = True
) -> str:
"""Prepare data to query string.

Expand Down Expand Up @@ -423,7 +422,7 @@ def slugify_string(
min_length: int = 1,
lower: bool = False,
make_set: bool = False,
) -> "Union[str, Set[str]]":
) -> Union[str, set[str]]:
"""Slugify a text string.

This function removes transliterates input string to ASCII, removes
Expand All @@ -443,7 +442,7 @@ def slugify_string(
min_length (int): Minimal length of an element (word).

Returns:
Union[str, Set[str]]: Based on 'make_set' value returns slugified
Union[str, set[str]]: Based on 'make_set' value returns slugified
string.

"""
Expand Down Expand Up @@ -473,8 +472,8 @@ def failed_json_default(value: Any) -> str:


def prepare_attribute_changes(
old_entity: "AnyEntityDict",
new_entity: "AnyEntityDict",
old_entity: AnyEntityDict,
new_entity: AnyEntityDict,
replace: int = False,
):
attrib_changes = {}
Expand Down Expand Up @@ -502,10 +501,10 @@ def prepare_attribute_changes(


def prepare_entity_changes(
old_entity: "AnyEntityDict",
new_entity: "AnyEntityDict",
old_entity: AnyEntityDict,
new_entity: AnyEntityDict,
replace: bool = False,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Prepare changes of entities."""
changes = {}
for key, new_value in new_entity.items():
Expand Down Expand Up @@ -537,7 +536,7 @@ def _try_parse_url(url: str) -> Optional[str]:
def _try_connect_to_server(
url: str,
timeout: Optional[float],
verify: Optional["Union[str, bool]"],
verify: Optional[Union[str, bool]],
cert: Optional[str],
) -> Optional[str]:
if timeout is None:
Expand Down Expand Up @@ -637,7 +636,7 @@ def get_user_by_token(
url: str,
token: str,
timeout: Optional[float] = None,
) -> Optional[Dict[str, Any]]:
) -> Optional[dict[str, Any]]:
"""Get user information by url and token.

Args:
Expand All @@ -647,7 +646,7 @@ def get_user_by_token(
'get_default_timeout' is used if not specified.

Returns:
Optional[Dict[str, Any]]: User information if url and token are valid.
Optional[dict[str, Any]]: User information if url and token are valid.

"""
if timeout is None:
Expand Down Expand Up @@ -699,7 +698,7 @@ def is_token_valid(
def validate_url(
url: str,
timeout: Optional[int] = None,
verify: Optional["Union[str, bool]"] = None,
verify: Optional[Union[str, bool]] = None,
cert: Optional[str] = None,
) -> str:
"""Validate url if is valid and server is available.
Expand Down Expand Up @@ -1146,7 +1145,7 @@ def get_media_mime_type_for_content(content: bytes) -> Optional[str]:
return _get_svg_mime_type(content)


def get_media_mime_type_for_stream(stream: "StreamType") -> Optional[str]:
def get_media_mime_type_for_stream(stream: StreamType) -> Optional[str]:
# Read only 12 bytes to determine mime type
content = stream.read(12)
if len(content) < 12:
Expand Down Expand Up @@ -1178,7 +1177,7 @@ def get_media_mime_type(filepath: str) -> Optional[str]:
def take_web_action_event(
server_url: str,
action_token: str
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Take web action event using action token.

Action token is generated by AYON server and passed to AYON launcher.
Expand All @@ -1188,7 +1187,7 @@ def take_web_action_event(
action_token (str): Action token.

Returns:
Dict[str, Any]: Web action event.
dict[str, Any]: Web action event.

"""
response = requests.get(
Expand Down