diff --git a/ayon_api/server_api.py b/ayon_api/server_api.py index f016c0b1f..aaa588973 100644 --- a/ayon_api/server_api.py +++ b/ayon_api/server_api.py @@ -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 @@ -87,7 +87,6 @@ ) if typing.TYPE_CHECKING: - from typing import Union from .typing import ( ServerVersion, AnyEntityDict, @@ -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) @@ -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, @@ -685,7 +684,7 @@ 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. @@ -693,7 +692,7 @@ def as_username( 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. @@ -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/). @@ -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 ) @@ -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: @@ -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]: @@ -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, @@ -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 @@ -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: @@ -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 diff --git a/ayon_api/utils.py b/ayon_api/utils.py index 7f04d64cd..fa9b7ac21 100644 --- a/ayon_api/utils.py +++ b/ayon_api/utils.py @@ -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 @@ -38,7 +38,6 @@ if typing.TYPE_CHECKING: - from typing import Union from .typing import AnyEntityDict, StreamType REMOVED_VALUE = object() @@ -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 @@ -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. @@ -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 @@ -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. """ @@ -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 = {} @@ -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(): @@ -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: @@ -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: @@ -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: @@ -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. @@ -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: @@ -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. @@ -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(