"""Server API.
Provides access to server API.
"""
from __future__ import annotations
import copy
from dataclasses import dataclass
import os
import re
import io
import json
import ssl
import time
import logging
import platform
import socket
import uuid
from contextlib import contextmanager
import typing
from typing import (
Iterable, Generator, Any, Literal, ContextManager
)
import requests
import websocket
from .constants import (
SERVER_RETRIES_ENV_KEY,
DEFAULT_FOLDER_TYPE_FIELDS,
DEFAULT_TASK_TYPE_FIELDS,
DEFAULT_PROJECT_STATUSES_FIELDS,
DEFAULT_PROJECT_TAGS_FIELDS,
DEFAULT_PRODUCT_TYPE_FIELDS,
DEFAULT_PROJECT_FIELDS,
DEFAULT_FOLDER_FIELDS,
DEFAULT_TASK_FIELDS,
DEFAULT_PRODUCT_FIELDS,
DEFAULT_VERSION_FIELDS,
DEFAULT_REPRESENTATION_FIELDS,
REPRESENTATION_FILES_FIELDS,
DEFAULT_WORKFILE_INFO_FIELDS,
DEFAULT_EVENT_FIELDS,
DEFAULT_ACTIVITY_FIELDS,
DEFAULT_USER_FIELDS,
DEFAULT_ENTITY_LIST_FIELDS,
DEFAULT_LINK_FIELDS,
)
from .graphql import INTROSPECTION_QUERY
from .graphql_queries import users_graphql_query
from .events import EventHub
from .exceptions import (
FailedOperations,
UnauthorizedError,
AuthenticationError,
ServerNotReached,
)
from .utils import (
RequestType,
RequestTypes,
RestApiResponse,
prepare_query_string,
logout_from_server,
create_entity_id,
entity_data_json_default,
failed_json_default,
TransferProgress,
get_default_timeout,
get_default_settings_variant,
get_default_site_id,
NOT_SET,
get_media_mime_type,
get_media_mime_type_for_stream,
get_machine_name,
fill_own_attribs,
UserInfo,
get_user_info_by_token,
)
from ._api_helpers import (
InstallersAPI,
DependencyPackagesAPI,
SecretsAPI,
BundlesAddonsAPI,
EventsAPI,
AttributesAPI,
ProjectsAPI,
FoldersAPI,
TasksAPI,
ProductsAPI,
VersionsAPI,
RepresentationsAPI,
WorkfilesAPI,
ThumbnailsAPI,
ActivitiesAPI,
ActionsAPI,
LinksAPI,
ListsAPI,
UsersAPI,
ServicesAPI,
SystemAPI,
ReviewablesAPI,
)
if typing.TYPE_CHECKING:
from websocket import WebSocket
from .typing import (
ServerVersion,
AnyEntityDict,
StreamType,
BackgroundOperationTask,
)
VERSION_REGEX = re.compile(
r"(?P<major>0|[1-9]\d*)"
r"\.(?P<minor>0|[1-9]\d*)"
r"\.(?P<patch>0|[1-9]\d*)"
r"(?:-(?P<prerelease>[a-zA-Z\d\-.]*))?"
r"(?:\+(?P<buildmetadata>[a-zA-Z\d\-.]*))?"
)
def _get_requests_ca_bundle() -> str | None:
"""CA bundle used by 'requests' for ssl verification.
Returns:
str | None: Path to CA bundle file or directory.
"""
# Same order as 'requests' uses when 'trust_env' is enabled
for env_key in ("REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE"):
path = os.environ.get(env_key)
if path:
return path
try:
from requests.certs import where
return where()
except Exception:
return None
def _disable_websocket_ipv6() -> None:
"""Skip IPv6 addresses in 'websocket' when python does not support IPv6.
Some DCC pythons are built without IPv6 support (e.g. Flame's on macOS).
'websocket' tries resolved addresses in order and fails on the first
IPv6 address with 'connect(): bad family' instead of trying the next
one. As 'localhost' resolves to '::1' first, websocket never
connected while REST requests (which use only IPv4) did work.
"""
if socket.has_ipv6:
return
try:
from websocket import _http
except ImportError:
return
# Private function of 'websocket', skip if not available
get_addrinfo_list = getattr(_http, "_get_addrinfo_list", None)
if get_addrinfo_list is None or getattr(
get_addrinfo_list, "_ayon_ipv6_disabled", False
):
return
def _get_ipv4_addrinfo_list(*args, **kwargs):
addrinfo_list, *other = get_addrinfo_list(*args, **kwargs)
ipv4_addrinfo_list = [
addrinfo
for addrinfo in addrinfo_list
if addrinfo[0] != socket.AF_INET6
]
return (ipv4_addrinfo_list or addrinfo_list, *other)
_get_ipv4_addrinfo_list._ayon_ipv6_disabled = True
_http._get_addrinfo_list = _get_ipv4_addrinfo_list
[docs]class GraphQlResponse:
"""GraphQl response."""
def __init__(self, data):
self.data = data
self.errors = data.get("errors")
def __len__(self):
if self.errors:
return 0
return 1
def __repr__(self):
if self.errors:
message = self.errors[0]["message"]
return f"<{self.__class__.__name__} errors={message}>"
return f"<{self.__class__.__name__}>"
class _AsUserStack:
"""Handle stack of users used over server api connection in service mode.
ServerAPI can behave as other users if it is using special API key.
Examples:
>>> stack = _AsUserStack()
>>> stack.set_default_username("DefaultName")
>>> print(stack.username)
DefaultName
>>> with stack.as_user("Other1"):
... print(stack.username)
... with stack.as_user("Other2"):
... print(stack.username)
... print(stack.username)
... stack.clear()
... print(stack.username)
Other1
Other2
Other1
None
>>> print(stack.username)
None
>>> stack.set_default_username("DefaultName")
>>> print(stack.username)
DefaultName
"""
def __init__(self):
self._users_by_id = {}
self._user_ids = []
self._last_user = None
self._default_user = None
def clear(self):
self._users_by_id = {}
self._user_ids = []
self._last_user = None
self._default_user = None
@property
def username(self) -> str | None:
# 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) -> str | None:
return self._default_user
def set_default_username(self, username: str | None = None) -> None:
self._default_user = username
default_username = property(get_default_username, set_default_username)
@contextmanager
def as_user(self, username: str | None) -> ContextManager[None]:
self._last_user = username
user_id = uuid.uuid4().hex
self._user_ids.append(user_id)
self._users_by_id[user_id] = username
def _cleanup():
self._users_by_id.pop(user_id, None)
if not self._user_ids:
return
# First check if is the user id the last one
was_last = self._user_ids[-1] == user_id
# Remove id from variables
if user_id in self._user_ids:
self._user_ids.remove(user_id)
if not was_last:
return
new_last_user = None
if self._user_ids:
new_last_user = self._users_by_id.get(self._user_ids[-1])
self._last_user = new_last_user
try:
yield
finally:
_cleanup()
[docs]@dataclass
class TokenInfo:
token: str | None = None
is_valid: bool | None = None
is_service: bool | None = None
username: str | None = None
unauthorized_response: requests.Response | None = None
[docs]class ServerAPI(
InstallersAPI,
DependencyPackagesAPI,
SecretsAPI,
BundlesAddonsAPI,
EventsAPI,
AttributesAPI,
ProjectsAPI,
FoldersAPI,
TasksAPI,
ProductsAPI,
VersionsAPI,
RepresentationsAPI,
WorkfilesAPI,
ThumbnailsAPI,
ActivitiesAPI,
ActionsAPI,
LinksAPI,
ListsAPI,
UsersAPI,
ServicesAPI,
SystemAPI,
ReviewablesAPI,
):
"""Base handler of connection to server.
Requires url to server which is used as base for api and graphql calls.
Login cause that a session is used
Args:
base_url (str): Example: http://localhost:5000
token (str | None): Access token (api key) to server.
site_id (str | None): Unique name of site. Should be the same when
connection is created from the same machine under same user.
client_version (str | None): Version of client application (used in
desktop client application).
default_settings_variant (Literal["production", "staging"] | None):
Settings variant used by default if a method for settings won't
get any (by default is 'production').
sender_type (str | None): Sender type of requests. Used in server
logs and propagated into events.
sender (str | None): Sender of requests, more specific than
sender type (e.g. machine name). Used in server logs and
propagated into events.
ssl_verify (bool | str | None): Verify SSL certificate
Looks for env variable value ``AYON_CA_FILE`` by default. If not
available then 'True' is used.
cert (str | None): Path to certificate file. Looks for env
variable value ``AYON_CERT_FILE`` by default.
create_session (bool): Create session for connection if
token is available. Default is True.
timeout (float | None): Timeout for requests.
max_retries (int | None): Number of retries for requests.
"""
_default_max_retries = 3
# 1 MB chunk by default
# TODO find out if these are reasonable default value
default_download_chunk_size = 1024 * 1024
default_upload_chunk_size = 1024 * 1024
def __init__(
self,
base_url: str,
token: str | None = None,
site_id: str | None = NOT_SET,
client_version: str | None = None,
default_settings_variant: str | None = None,
sender_type: str | None = None,
sender: str | None = None,
ssl_verify: bool | str | None = None,
cert: str | None = None,
create_session: bool = True,
timeout: float | None = None,
max_retries: int | None = None,
) -> None:
if not base_url:
raise ValueError(f"Invalid server URL {str(base_url)}")
base_url = base_url.rstrip("/")
self._base_url: str = base_url
self._rest_url: str = f"{base_url}/api"
self._graphql_url: str = f"{base_url}/graphql"
self._log: logging.Logger = logging.getLogger(self.__class__.__name__)
# Allow to have 'site_id' to 'None'
if site_id is NOT_SET:
site_id = get_default_site_id()
self._site_id: str | None = site_id
self._client_version: str | None = client_version
self._default_settings_variant: str = (
default_settings_variant
or get_default_settings_variant()
)
self._sender: str | None = sender
self._sender_type: str | None = sender_type
self._timeout: float = 0.0
self._max_retries: int = 0
# Set timeout and max retries based on passed values
self.set_timeout(timeout)
self.set_max_retries(max_retries)
if ssl_verify is None:
# Custom AYON env variable for CA file or 'True'
# - that should cover most default behaviors in 'requests'
# with 'certifi'
ssl_verify = os.environ.get("AYON_CA_FILE") or True
if cert is None:
cert = os.environ.get("AYON_CERT_FILE")
self._ssl_verify = ssl_verify
self._cert = cert
self._token_info: TokenInfo = TokenInfo(token=token)
self._server_available = None
self._server_version = None
self._server_version_tuple = None
self._graphql_allows_traits_in_representations: bool | None = None
self._product_base_type_supported: bool | None = None
self._links_graphql_support_data: bool | None = None
self._session = None
# Session without authorization used for availability check and
# token validation, see '_get_validation_session'
self._validation_session: requests.Session | None = None
self._base_functions_mapping = {
RequestTypes.get: requests.get,
RequestTypes.post: requests.post,
RequestTypes.put: requests.put,
RequestTypes.patch: requests.patch,
RequestTypes.delete: requests.delete
}
self._session_functions_mapping = {}
# Attributes cache
self._attributes_schema = None
self._entity_type_attributes_cache = {}
self._as_user_stack: _AsUserStack = _AsUserStack()
# Create session
if self._token_info.token and create_session:
self.validate_server_availability()
self.create_session()
# Created on demand by 'get_event_hub'
self._event_hub: EventHub | None = None
@property
def log(self) -> logging.Logger:
return self._log
[docs] def get_base_url(self) -> str:
return self._base_url
[docs] def get_rest_url(self) -> str:
return self._rest_url
[docs] def get_websocket_url(
self, endpoint: str
) -> str:
"""Prepare websocket url from endpoint."""
endpoint = (endpoint or "").strip()
if self._base_url.startswith("https"):
base_url = f"wss{self._base_url[5:]}"
elif self._base_url.startswith("http"):
base_url = f"ws{self._base_url[4:]}"
else:
raise ValueError(f"Invalid scheme in base URL: {self._base_url}")
base_url = base_url.rstrip("/")
endpoint = endpoint.lstrip("/")
return f"{base_url}/{endpoint}"
base_url = property(get_base_url)
rest_url = property(get_rest_url)
[docs] def get_ssl_verify(self) -> bool | str | None:
"""Enable ssl verification.
Returns:
bool | str | None: Current state of ssl verification.
"""
return self._ssl_verify
[docs] def set_ssl_verify(
self, ssl_verify: bool | str | None
) -> None:
"""Change ssl verification state.
Args:
ssl_verify (bool | str | None): Enabled/disable
ssl verification, can be a path to file.
"""
if self._ssl_verify == ssl_verify:
return
self._ssl_verify = ssl_verify
for session in (self._session, self._validation_session):
if session is not None:
session.verify = ssl_verify
[docs] def get_cert(self) -> str | None:
"""Current cert file used for connection to server.
Returns:
str | None: Path to cert file.
"""
return self._cert
[docs] def set_cert(self, cert: str | None) -> None:
"""Change cert file used for connection to server.
Args:
cert (str | None): Path to cert file.
"""
if cert == self._cert:
return
self._cert = cert
for session in (self._session, self._validation_session):
if session is not None:
session.cert = cert
ssl_verify = property(get_ssl_verify, set_ssl_verify)
cert = property(get_cert, set_cert)
[docs] @classmethod
def get_default_timeout(cls) -> float:
"""Default value for requests timeout.
Utils function 'get_default_timeout' is used by default.
Returns:
float: Timeout value in seconds.
"""
return get_default_timeout()
[docs] @classmethod
def get_default_max_retries(cls) -> int:
"""Default value for requests max retries.
First looks for environment variable SERVER_RETRIES_ENV_KEY, which
can affect max retries value. If not available then use class
attribute '_default_max_retries'.
Returns:
int: Max retries value.
"""
try:
return int(os.environ.get(SERVER_RETRIES_ENV_KEY))
except (ValueError, TypeError):
pass
return cls._default_max_retries
[docs] def get_timeout(self) -> float:
"""Current value for requests timeout.
Returns:
float: Timeout value in seconds.
"""
return self._timeout
[docs] def set_timeout(self, timeout: int | float | None) -> None:
"""Change timeout value for requests.
Args:
timeout (float | None): Timeout value in seconds.
"""
if timeout is None:
timeout = self.get_default_timeout()
self._timeout = float(timeout)
[docs] def get_max_retries(self) -> int:
"""Current value for requests max retries.
Returns:
int: Max retries value.
"""
return self._max_retries
[docs] def set_max_retries(self, max_retries: int | None) -> None:
"""Change max retries value for requests.
Args:
max_retries (int | None): Max retries value.
"""
if max_retries is None:
max_retries = self.get_default_max_retries()
self._max_retries = int(max_retries)
timeout = property(get_timeout, set_timeout)
max_retries = property(get_max_retries, set_max_retries)
@property
def access_token(self) -> str | None:
"""Access token used for authorization to server.
Returns:
str | None: Token string or None if not authorized yet.
"""
return self._token_info.token
@property
def access_token_username(self) -> str | None:
"""Name of user who owns the access token.
The value is stored on token validation, so it does not require
additional request. It is not affected by 'as_username', use
'get_user' to get user the requests are made as.
Returns:
str | None: Username or None if token is not valid.
"""
if not self.has_valid_token:
return None
if self._token_info.username is None:
self.validate_token()
return self._token_info.username
[docs] def is_service_user(self) -> bool:
"""Check if connection is using service API key.
Returns:
bool: Used api key belongs to service user.
"""
if not self.has_valid_token:
raise ValueError("User is not logged in.")
return bool(self._token_info.is_service)
[docs] def get_site_id(self) -> str | None:
"""Site id used for connection.
Site id tells server from which machine/site is connection created and
is used for default site overrides when settings are received.
Returns:
str | None: Site id value or None if not filled.
"""
return self._site_id
[docs] def set_site_id(self, site_id: str | None) -> None:
"""Change site id of connection.
Behave as specific site for server. It affects default behavior of
settings getter methods.
Args:
site_id (str | None): Site id value, or 'None' to unset.
"""
if self._site_id == site_id:
return
self._site_id = site_id
# Recreate session on machine id change
self._update_session_headers()
site_id = property(get_site_id, set_site_id)
[docs] def get_client_version(self) -> str | None:
"""Version of client used to connect to server.
Client version is AYON client build desktop application.
Returns:
str: Client version string used in connection.
"""
return self._client_version
[docs] def set_client_version(self, client_version: str | None) -> None:
"""Set version of client used to connect to server.
Client version is AYON client build desktop application.
Args:
client_version (str | None): Client version string.
"""
if self._client_version == client_version:
return
self._client_version = client_version
self._update_session_headers()
client_version = property(get_client_version, set_client_version)
[docs] def get_default_settings_variant(self) -> str | None:
"""Default variant used for settings.
Returns:
str | None: name of variant or None.
"""
return self._default_settings_variant
[docs] def set_default_settings_variant(self, variant: str) -> None:
"""Change default variant for addon settings.
Note:
It is recommended to set only 'production' or 'staging' variants
as default variant.
Args:
variant (str): Settings variant name. It is possible to use
'production', 'staging' or name of dev bundle.
"""
self._default_settings_variant = variant
default_settings_variant = property(
get_default_settings_variant,
set_default_settings_variant
)
[docs] def get_sender(self) -> str | None:
"""Sender used to send requests.
Returns:
str | None: Sender name or None.
"""
return self._sender
[docs] def set_sender(self, sender: str | None) -> None:
"""Change sender used for requests.
Args:
sender (str | None): Sender name or None.
"""
if sender == self._sender:
return
self._sender = sender
self._update_session_headers()
sender = property(get_sender, set_sender)
[docs] def get_sender_type(self) -> str | None:
"""Sender type used to send requests.
Sender type is supported since AYON server 1.5.5 .
Returns:
str | None: Sender type or None.
"""
return self._sender_type
[docs] def set_sender_type(self, sender_type: str | None) -> None:
"""Change sender type used for requests.
Args:
sender_type (str | None): Sender type or None.
"""
if sender_type == self._sender_type:
return
self._sender_type = sender_type
self._update_session_headers()
sender_type = property(get_sender_type, set_sender_type)
[docs] def get_default_service_username(self) -> str | None:
"""Default username used for callbacks when used with service API key.
Returns:
str | None: Username if any was filled.
"""
return self._as_user_stack.get_default_username()
[docs] def set_default_service_username(
self, username: str | None = None
) -> None:
"""Service API will work as other user.
Service API keys can work as other user. It can be temporary using
context manager 'as_user' or it is possible to set default username if
'as_user' context manager is not entered.
Args:
username (str | None): Username to work as when service.
Raises:
ValueError: When connection is not yet authenticated or api key
is not service token.
"""
current_username = self._as_user_stack.get_default_username()
if current_username == username:
return
if not self.has_valid_token:
raise ValueError(
"Authentication of connection did not happen yet."
)
if not self._token_info.is_service:
raise ValueError(
"Can't set service username. API key is not a service token."
)
self._as_user_stack.set_default_username(username)
if self._as_user_stack.username == username:
self._update_session_headers()
[docs] @contextmanager
def as_username(
self,
username: str | None,
ignore_service_error: bool = False,
) -> ContextManager[None]:
"""Service API will temporarily work as other user.
This method can be used only if service API key is logged in.
Args:
username (str | None): Username to work as when service.
ignore_service_error (bool): Ignore error when service
API key is not used.
Raises:
ValueError: When connection is not yet authenticated or api key
is not service token.
"""
if not self.has_valid_token:
raise ValueError(
"Authentication of connection did not happen yet."
)
if not self._token_info.is_service:
if ignore_service_error:
yield
return
raise ValueError(
"Can't set service username. API key is not a service token."
)
try:
with self._as_user_stack.as_user(username):
self._update_session_headers()
yield
finally:
self._update_session_headers()
@property
def is_server_available(self) -> bool:
if (
self._server_available is None
and self._token_info.token
and self._token_info.is_valid is None
):
# Token validation also finds out if server is available,
# which saves a request
user_info, responded = self._request_token_user_info()
if responded:
self._store_token_validation(user_info)
if self._server_available is None:
response = self._get_validation_session().get(
f"{self._rest_url}/info",
timeout=self.timeout,
)
self._server_available = response.status_code == 200
return self._server_available
@property
def has_valid_token(self) -> bool:
if self._token_info.token is None:
return False
if self._token_info.is_valid is None:
self.validate_token()
return self._token_info.is_valid
[docs] def validate_server_availability(self) -> None:
if not self.is_server_available:
raise ServerNotReached(
f"Server \"{self._base_url}\" can't be reached"
)
[docs] def validate_token(self) -> bool:
if self._token_info.token is None:
self._token_info.is_valid = False
self._token_info.unauthorized_response = None
self.close_session()
return False
user_info, responded = self._request_token_user_info()
if not responded:
# Check server url, raises an error if the url does not lead
# to AYON server
self._get_server_info()
self._store_token_validation(user_info)
return self._token_info.is_valid
def _request_token_user_info(self) -> tuple[UserInfo, bool]:
"""Request information about user of the token.
Server is marked as available if it did respond to the token.
Returns:
tuple[UserInfo, bool]: User information and if server did
respond with success or unauthorized status.
"""
user_info = get_user_info_by_token(
self.base_url,
self._token_info.token,
verify=self._ssl_verify,
cert=self._cert,
timeout=self.timeout,
session=self._get_validation_session(),
)
responded = (
user_info.response is not None
and user_info.response.status_code in (200, 401)
)
if responded:
self._server_available = True
return user_info, responded
def _store_token_validation(self, user_info: UserInfo) -> None:
self._token_info.is_valid = user_info.is_valid
self._token_info.unauthorized_response = user_info.response
is_service = username = None
if user_info.is_valid:
is_service = user_info.is_service
username = user_info.username
self._token_info.is_service = is_service
self._token_info.username = username
[docs] def set_token(self, token: str | None) -> None:
self._reset_token()
self._token_info.token = token
self.validate_token()
self._token_changed()
[docs] def reset_token(self) -> None:
self._reset_token()
self._token_changed()
def _reset_token(self) -> None:
self._token_info.token = None
self._token_info.is_service = None
self._token_info.username = None
self._token_info.is_valid = None
self._token_info.unauthorized_response = None
self.close_session()
[docs] def create_session(
self, ignore_existing: bool = True, force: bool = False
) -> None:
"""Create a connection session.
Session helps to keep connection with server without
need to reconnect on each call.
Args:
ignore_existing (bool): If session already exists,
ignore creation.
force (bool): If session already exists, close it and
create new.
"""
if force and self._session is not None:
self.close_session()
if self._session is not None:
if ignore_existing:
return
raise ValueError("Session is already created.")
self._as_user_stack.clear()
# Validate token before session creation, unless it was already
# validated (e.g. by server availability check)
if self._token_info.is_valid is None:
self.validate_token()
# Reuse connection of availability check and token validation
session = self._validation_session
self._validation_session = None
if session is None:
session = requests.Session()
session.cert = self._cert
session.verify = self._ssl_verify
session.headers.update(self.get_headers())
self._session_functions_mapping = {
RequestTypes.get: session.get,
RequestTypes.post: session.post,
RequestTypes.put: session.put,
RequestTypes.patch: session.patch,
RequestTypes.delete: session.delete
}
self._session = session
[docs] def close_session(self) -> None:
if self._validation_session is not None:
self._validation_session.close()
self._validation_session = None
if self._session is None:
return
session = self._session
self._session = None
self._session_functions_mapping = {}
session.close()
def _get_validation_session(self) -> requests.Session:
"""Session for requests made before authorized session is created.
Server availability check and token validation use the same
session, so they share one connection. The session is then used
as the authorized session by 'create_session', which reuses
the connection too.
Returns:
requests.Session: Session without authorization headers.
"""
if self._validation_session is None:
session = requests.Session()
session.cert = self._cert
session.verify = self._ssl_verify
self._validation_session = session
return self._validation_session
[docs] def create_websocket(
self,
endpoint: str,
*,
timeout: float | None = None,
headers: dict[str, Any] | None = None,
sslopt: dict[str, Any] | None = None,
**kwargs,
) -> WebSocket:
"""Create a websocket connection to AYON server."""
ws_url = self.get_websocket_url(endpoint)
ws_headers = self.get_headers()
ws_headers.pop("Content-Type", None)
if headers:
ws_headers.update(headers)
ws_kwargs = copy.deepcopy(kwargs)
if timeout is None:
timeout = self.timeout
if timeout:
ws_kwargs["timeout"] = timeout
if ws_headers:
ws_kwargs["header"] = [
f"{key}: {value}"
for key, value in ws_headers.items()
if value is not None
]
prepared_sslopt = copy.deepcopy(sslopt) if sslopt else {}
if self._ssl_verify is False:
prepared_sslopt.setdefault("cert_reqs", ssl.CERT_NONE)
else:
# Verify with the same CA bundle as REST requests do. Without
# it 'websocket' uses system certificates of the python ssl
# module, which might be missing (e.g. python on macOS), so
# REST would work but websocket would always fail.
ca_bundle = self._ssl_verify
if not isinstance(ca_bundle, str):
ca_bundle = _get_requests_ca_bundle()
if ca_bundle:
key = "ca_certs"
if os.path.isdir(ca_bundle):
key = "ca_cert_path"
prepared_sslopt.setdefault(key, ca_bundle)
if self._cert:
prepared_sslopt.setdefault("certfile", self._cert)
if ws_url.startswith("wss://") and prepared_sslopt:
ws_kwargs["sslopt"] = prepared_sslopt
_disable_websocket_ipv6()
return websocket.create_connection(ws_url, **ws_kwargs)
def _update_session_headers(self) -> None:
if self._session is None:
return
# Header keys that may change over time
for key, value in (
("X-as-user", self._as_user_stack.username),
("x-ayon-version", self._client_version),
("x-ayon-site-id", self._site_id),
("x-sender-type", self._sender_type),
("x-sender", self._sender),
):
if value is not None:
self._session.headers[key] = value
elif key in self._session.headers:
self._session.headers.pop(key)
[docs] def get_info(self) -> dict[str, Any]:
"""Get information about current used api key.
By default, the 'info' contains only 'uptime' and 'version'. With
logged user info also contains information about user and machines on
which was logged in.
Todos:
Use this method for validation of token instead of 'get_user'.
Returns:
dict[str, Any]: Information from server.
"""
handle_invalid_token = (
self._token_info.token
and self._token_info.is_valid
)
response = self.raw_get(
"info",
handle_invalid_token=handle_invalid_token,
)
response.raise_for_status()
return response.data
[docs] def get_server_version(self) -> str:
"""Get server version.
Version should match semantic version (https://semver.org/).
Returns:
str: Server version.
"""
if self._server_version is None:
self._server_version = self.get_info()["version"]
return self._server_version
[docs] def get_server_version_tuple(self) -> ServerVersion:
"""Get server version as tuple.
Version should match semantic version (https://semver.org/).
This function only returns first three numbers of version.
Returns:
ServerVersion: Server version.
"""
if self._server_version_tuple is None:
re_match = VERSION_REGEX.fullmatch(
self.get_server_version())
self._server_version_tuple = (
int(re_match.group("major")),
int(re_match.group("minor")),
int(re_match.group("patch")),
re_match.group("prerelease") or "",
re_match.group("buildmetadata") or "",
)
return self._server_version_tuple
server_version = property(get_server_version)
server_version_tuple: ServerVersion = property(
get_server_version_tuple
)
@property
def graphql_allows_traits_in_representations(self) -> bool:
"""Check server support for representation traits."""
if self._graphql_allows_traits_in_representations is None:
major, minor, patch, _, _ = self.server_version_tuple
self._graphql_allows_traits_in_representations = (
(major, minor, patch) >= (1, 7, 5)
)
return self._graphql_allows_traits_in_representations
[docs] def is_product_base_type_supported(self) -> bool:
"""Product base types are available on server."""
if self._product_base_type_supported is None:
major, minor, patch, _, _ = self.server_version_tuple
self._product_base_type_supported = (
(major, minor, patch) >= (1, 13, 0)
)
return self._product_base_type_supported
[docs] def links_graphql_support_data(self) -> bool:
"""Links data can be received by GraphQl."""
if self._links_graphql_support_data is None:
major, minor, patch, _, _ = self.server_version_tuple
self._links_graphql_support_data = (
(major, minor, patch) >= (1, 14, 2)
)
return self._links_graphql_support_data
[docs] def get_users(
self,
project_name: str | None = None,
usernames: Iterable[str] | None = None,
emails: Iterable[str] | None = None,
fields: Iterable[str] | None = None,
own_attributes: bool = False,
) -> Generator[dict[str, Any], None, None]:
"""Get Users.
Only administrators and managers can fetch all users. For other users
it is required to pass in 'project_name' filter.
Args:
project_name (str | None): Project name.
usernames (Iterable[str] | None): Filter by usernames.
emails (Iterable[str] | None): Filter by emails.
fields (Iterable[str] | None): Fields to be queried
for users.
own_attributes (bool): Fill 'own_attributes' with
attribute values that are explicitly set on user, other
attributes have 'None' value.
Returns:
Generator[dict[str, Any], None, None]: Queried users.
"""
filters = {}
if usernames is not None:
usernames = set(usernames)
if not usernames:
return
filters["userNames"] = list(usernames)
if emails is not None:
emails = set(emails)
if not emails:
return
major, minor, patch, _, _ = self.server_version_tuple
emails_filter_available = (major, minor, patch) > (1, 7, 3)
if not emails_filter_available:
server_version = self.get_server_version()
raise ValueError(
"Filtering by emails is not supported by"
f" server version {server_version}."
)
filters["emails"] = list(emails)
if project_name is not None:
filters["projectName"] = project_name
if not fields:
fields = self.get_default_fields_for_type("user")
else:
fields = set(fields)
self._prepare_fields("user", fields)
query = users_graphql_query(fields)
for attr, filter_value in filters.items():
query.set_variable_value(attr, filter_value)
attributes = {}
if "allAttrib" in fields:
attributes = self.get_attributes_for_type("user")
for parsed_data in query.continuous_query(self):
for user in parsed_data["users"]:
access_groups = user.get("accessGroups")
if isinstance(access_groups, str):
user["accessGroups"] = json.loads(access_groups)
attrib = user.get("allAttrib")
if isinstance(attrib, str):
attrib = json.loads(attrib)
if attrib is not None:
own_attrib = copy.deepcopy(attrib)
user["ownAttrib"] = own_attrib
for name, attr_data in attributes.items():
attrib.setdefault(name, attr_data["default"])
own_attrib.setdefault(name, None)
user["attrib"] = attrib
if own_attributes:
user["own_attributes"] = copy.deepcopy(own_attrib)
yield user
[docs] def get_user_by_name(
self,
username: str,
project_name: str | None = None,
fields: Iterable[str] | None = None,
own_attributes: bool = False,
) -> dict[str, Any] | None:
"""Get user by name using GraphQl.
Only administrators and managers can fetch all users. For other users
it is required to pass in 'project_name' filter.
Args:
username (str): Username.
project_name (str | None): Define scope of project.
fields (Iterable[str] | None): Fields to be queried for users.
own_attributes (bool): Fill 'own_attributes' with
attribute values that are explicitly set on user, other
attributes have 'None' value.
Returns:
dict[str, Any] | None: User info or None if user is not found.
"""
if not username:
return None
for user in self.get_users(
project_name=project_name,
usernames={username},
fields=fields,
own_attributes=own_attributes,
):
return user
return None
[docs] def get_user(
self,
username: str | None = None,
own_attributes: bool = False,
fill_attrib: bool = True,
) -> dict[str, Any] | None:
"""Get user info using REST endpoint.
User contains only explicitly set attributes in 'attrib'.
Args:
username (str | None): Username.
own_attributes (bool): Fill 'own_attributes' with
attribute values that are explicitly set on user, other
attributes have 'None' value.
fill_attrib (bool): Fill missing attributes in 'attrib' with their
default values. Requires an additional request for user
attributes. Use 'False' when attributes are not needed.
Returns:
dict[str, Any] | None: User info or None if user is not
found.
"""
if username is None:
user = self._get_user_info()
if user is None:
raise UnauthorizedError("User is not authorized.")
else:
response = self.get(f"users/{username}")
response.raise_for_status()
user = response.data
if fill_attrib:
attributes = self.get_attributes_for_type("user")
for attr_name, attr_data in attributes.items():
user["attrib"].setdefault(attr_name, attr_data["default"])
# NOTE 'ownAttrib' is always filled for backwards compatibility
fill_own_attribs(user)
if not own_attributes:
user.pop("own_attributes", None)
return user
[docs] def get_event_hub(self) -> EventHub:
"""Get event hub for connection.
Event hub is used to subscribe to events from server. The hub is
created on first call, websocket connection is created when
a callback is registered or when the hub is started.
Notes:
Running hub keeps reference to the connection, call
'EventHub.stop' when the connection is not used anymore.
Returns:
EventHub: Event hub instance.
"""
if self._event_hub is None:
self._event_hub = EventHub(self)
return self._event_hub
def _token_changed(self) -> None:
"""Let event hub know that token did change."""
if self._event_hub is not None:
self._event_hub.reauthenticate()
[docs] def login(
self,
username: str,
password: str,
create_session: bool = True,
) -> None:
"""Login to server.
Args:
username (str): Username.
password (str): Password.
create_session (bool): Create session after login.
Default: True.
Raises:
AuthenticationError: Login failed.
"""
if self.has_valid_token:
if self.access_token_username == username:
self.close_session()
if create_session:
self.create_session()
return
self._reset_token()
try:
self.validate_server_availability()
response = self.raw_post(
"auth/login",
json=dict(
name=username,
password=password,
),
handle_invalid_token=False,
)
if response.status_code != 200:
_detail = response.data.get("detail")
details = ""
if _detail:
details = f" {_detail}"
raise AuthenticationError(f"Login failed {details}")
self._token_info.token = response["token"]
# Should be valid if was just logged in
self._token_info.is_valid = True
# Service token can't be obtained by login, so it is not
# service token
self._token_info.is_service = False
user = response.data.get("user") or {}
self._token_info.username = user.get("name")
if not self.has_valid_token:
raise AuthenticationError("Invalid credentials")
finally:
self._token_changed()
if create_session:
self.create_session()
[docs] def logout(self, soft: bool = False) -> None:
if self._token_info.token:
if not soft:
self._logout()
self.reset_token()
[docs] def raw_post(self, entrypoint: str, **kwargs) -> RestApiResponse:
url = self._endpoint_to_url(entrypoint)
self.log.debug(f"Executing [POST] {url}")
return self._do_rest_request(
RequestTypes.post,
url,
**kwargs
)
[docs] def raw_put(self, entrypoint: str, **kwargs) -> RestApiResponse:
url = self._endpoint_to_url(entrypoint)
self.log.debug(f"Executing [PUT] {url}")
return self._do_rest_request(
RequestTypes.put,
url,
**kwargs
)
[docs] def raw_patch(self, entrypoint: str, **kwargs) -> RestApiResponse:
url = self._endpoint_to_url(entrypoint)
self.log.debug(f"Executing [PATCH] {url}")
return self._do_rest_request(
RequestTypes.patch,
url,
**kwargs
)
[docs] def raw_get(self, entrypoint: str, **kwargs) -> RestApiResponse:
url = self._endpoint_to_url(entrypoint)
self.log.debug(f"Executing [GET] {url}")
return self._do_rest_request(
RequestTypes.get,
url,
**kwargs
)
[docs] def raw_delete(self, entrypoint: str, **kwargs) -> RestApiResponse:
url = self._endpoint_to_url(entrypoint)
self.log.debug(f"Executing [DELETE] {url}")
return self._do_rest_request(
RequestTypes.delete,
url,
**kwargs
)
[docs] def post(self, entrypoint: str, **kwargs) -> RestApiResponse:
return self.raw_post(entrypoint, json=kwargs)
[docs] def put(self, entrypoint: str, **kwargs) -> RestApiResponse:
return self.raw_put(entrypoint, json=kwargs)
[docs] def patch(self, entrypoint: str, **kwargs) -> RestApiResponse:
return self.raw_patch(entrypoint, json=kwargs)
[docs] def get(self, entrypoint: str, **kwargs) -> RestApiResponse:
return self.raw_get(entrypoint, params=kwargs)
[docs] def delete(self, entrypoint: str, **kwargs) -> RestApiResponse:
return self.raw_delete(entrypoint, params=kwargs)
[docs] def get_server_config(self) -> dict[str, Any]:
response = self.get("config")
response.raise_for_status()
return response.data
[docs] def set_server_config(
self,
studio_name: str | None = None,
customization: dict[str, Any] | None = None,
authentication: dict[str, Any] | None = None,
project_options: dict[str, Any] | None = None,
changelog: dict[str, Any] | None = None,
) -> None:
body = {
key: value
for key, value in (
("studio_name", studio_name),
("customization", customization),
("authentication", authentication),
("project_options", project_options),
("changelog", changelog),
)
if value is not None
}
response = self.post("config", **body)
response.raise_for_status()
[docs] def get_server_config_overrides(self) -> dict[str, Any]:
response = self.get("config/overrides")
response.raise_for_status()
return response.data
[docs] def get_server_config_value(self, key: str) -> Any:
response = self.get(f"config/value/{key}")
response.raise_for_status()
return response.data
[docs] def download_server_config_file(
self,
file_type: Literal["login_background", "studio_logo"],
filepath: str,
*,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> TransferProgress:
"""Download server config file.
Validate if server has config file available first. Method crashes
if the file is not available.
Args:
file_type (Literal["login_background", "studio_logo"]): File to
download.
filepath (str): Target filepath.
chunk_size (int | None): Size of chunks used for download.
progress (TransferProgress | None): Object to track download
progress.
"""
return self.download_file(
f"api/config/files/{file_type}",
filepath,
chunk_size=chunk_size,
progress=progress,
)
[docs] def download_server_config_file_to_stream(
self,
file_type: Literal["login_background", "studio_logo"],
stream: StreamType,
*,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> TransferProgress:
"""Download server config file to byte stream.
Validate if server has config file available first. Method crashes
if the file is not available.
Args:
file_type (Literal["login_background", "studio_logo"]): File to
download.
stream (StreamType): Stream where downloaded content is stored.
chunk_size (int | None): Size of chunks used for download.
progress (TransferProgress | None): Object to track download
progress.
"""
return self.download_file_to_stream(
f"api/config/files/{file_type}",
stream,
chunk_size=chunk_size,
progress=progress,
)
[docs] def upload_server_config_file(
self,
file_type: Literal["login_background", "studio_logo"],
filepath: str,
*,
content_type: str | None = None,
filename: str | None = None,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> requests.Response:
"""Upload server config file from byte stream.
TODO create filename using file_type and extension from content_type
if filename is not specified
Args:
file_type (Literal["login_background", "studio_logo"]): File to
download.
filepath (str): Filepath used to store the file.
chunk_size (int | None): Size of chunks used for download.
progress (TransferProgress | None): Object to track download
progress.
Returns:
requests.Response: Response from upload.
"""
if not filename:
filename = os.path.basename(filepath)
return self.upload_file(
f"api/config/files/{file_type}",
filepath,
filename=filename,
content_type=content_type,
chunk_size=chunk_size,
progress=progress,
)
[docs] def upload_server_config_file_from_stream(
self,
file_type: Literal["login_background", "studio_logo"],
stream: StreamType,
filename: str,
*,
content_type: str | None = None,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> requests.Response:
"""Upload server config file from byte stream.
TODO create filename using file_type and extension from content_type
if filename is not specified
Args:
file_type (Literal["login_background", "studio_logo"]): File to
download.
stream (StreamType): Stream where downloaded content is stored.
filename (str): Filename used to store the file.
chunk_size (int | None): Size of chunks used for download.
progress (TransferProgress | None): Object to track download
progress.
Returns:
requests.Response: Response from upload.
"""
return self.upload_file_from_stream(
f"api/config/files/{file_type}",
stream,
filename=filename,
content_type=content_type,
chunk_size=chunk_size,
progress=progress,
)
def _endpoint_to_url(
self,
endpoint: str,
use_rest: bool = True,
) -> str:
"""Cleanup endpoint and return full url to AYON server.
If endpoint already starts with server url only slashes are removed.
Args:
endpoint (str): Endpoint to be cleaned.
use_rest (bool): Use only base server url if set to
False, otherwise REST endpoint is used.
Returns:
str: Full url to AYON server.
"""
endpoint = endpoint.lstrip("/").rstrip("/")
if endpoint.startswith(self._base_url):
return endpoint
base_url = self._rest_url if use_rest else self._base_url
return f"{base_url}/{endpoint}"
def _logout(self) -> None:
if self._token_info.is_valid:
logout_from_server(self._base_url, self._token_info.token)
def _get_server_info(self) -> dict[str, Any]:
"""Get server info without a session."""
response = self.raw_get(
"info",
handle_invalid_token=False,
verify=self.ssl_verify,
cert=self.cert
)
response.raise_for_status()
return response.data
def _get_user_info(self) -> dict[str, Any] | None:
if (
self._token_info.token is None
or self._token_info.is_valid is False
):
return None
if self._token_info.is_service is None:
self.validate_token()
if self._token_info.is_valid is False:
return None
response = self.get("users/me")
if response.status == 200:
return response.data
return None
def _do_rest_request(
self,
function: Any,
url: str,
*,
handle_invalid_token: bool = True,
**kwargs
) -> RestApiResponse:
kwargs.setdefault("timeout", self.timeout)
max_retries = kwargs.pop("max_retries", self.max_retries)
if max_retries < 1:
max_retries = 1
if handle_invalid_token and self._token_info.is_valid is False:
# Return a fake error response if the token is known to be invalid.
# Added to prevent DDOS attack on server when many requests
# with invalid token are send. It is better to return error
# immediately without trying to send a request to server.
if self._token_info.unauthorized_response is not None:
return RestApiResponse(self._token_info.unauthorized_response)
new_response = RestApiResponse(
None,
{"code": 401, "detail": "AYON api error: Invalid API key"}
)
new_response.status = 401
return new_response
if self._session is None:
# Validate token if was not yet validated
if (
handle_invalid_token
and self._token_info.is_valid is None
):
self.validate_token()
if "headers" not in kwargs:
kwargs["headers"] = self.get_headers()
if isinstance(function, RequestType):
function = self._base_functions_mapping[function]
elif isinstance(function, RequestType):
function = self._session_functions_mapping[function]
response = None
new_response = None
for retry_idx in reversed(range(max_retries)):
new_response = None
try:
response = function(url, **kwargs)
# Usually these mean, try later.
# 502: returned by the proxy: nginx
# 503: returned by the server: if no capacity
if response.status_code in {502, 503}:
new_response = RestApiResponse(response)
self.log.warning(
"Server returned %s status code."
" Retrying with longer delay...",
response.status_code
)
if retry_idx != 0:
time.sleep(2)
continue
break
except ConnectionRefusedError:
if retry_idx == 0:
self.log.warning(
"AYON api error: Connection error happened.",
exc_info=True,
)
# Server may be restarting
new_response = RestApiResponse(
None,
{
"detail": (
"AYON api error: Unable to connect the server."
" Connection refused"
)
}
)
except requests.exceptions.Timeout:
# Connection timed out
new_response = RestApiResponse(
None,
{"detail": "AYON api error: Connection timed out."}
)
except requests.exceptions.ConnectionError:
# Log warning only on last attempt
if retry_idx == 0:
self.log.warning(
"AYON api error: Connection error happened.",
exc_info=True
)
new_response = RestApiResponse(
None,
{
"detail": (
"AYON api error: Unable to connect the server."
" Connection error."
)
}
)
if retry_idx != 0:
time.sleep(0.1)
if new_response is not None:
return new_response
new_response = RestApiResponse(response)
if (
handle_invalid_token
and new_response.status_code == 401
and self._token_info.is_valid
):
self._token_info.is_valid = False
self._token_info.unauthorized_response = response
self.close_session()
self.log.debug(f"Response {str(new_response)}")
return new_response
def _download_file_to_stream(
self,
endpoint: str,
stream: StreamType,
chunk_size: int,
progress: TransferProgress,
) -> None:
headers = self.get_headers()
kwargs = {
"stream": True,
"headers": headers,
}
if self._session is None:
get_func = self._base_functions_mapping[RequestTypes.get]
else:
get_func = self._session_functions_mapping[RequestTypes.get]
url = self._endpoint_to_url(endpoint, use_rest=False)
progress.set_source_url(url)
retries = max(self.max_retries, 1)
api_prepended = False
for attempt in range(retries):
# Continue in download
offset = progress.get_transferred_size()
if offset > 0:
headers["Range"] = f"bytes={offset}-"
try:
with get_func(url, **kwargs) as response:
# Auto-fix missing 'api/'
if (
response.status_code in (404, 405)
and not api_prepended
):
api_prepended = True
if (
not endpoint.startswith(self._base_url)
and not endpoint.startswith("api/")
):
url = self._endpoint_to_url(
endpoint, use_rest=True
)
progress.set_destination_url(url)
continue
response.raise_for_status()
if offset > 0 and response.status_code != 206:
# Server ignored 'Range' and sends whole file again,
# already downloaded content must be discarded
stream.seek(0)
stream.truncate()
progress.reset_transferred()
headers.pop("Range", None)
if progress.get_content_size() is None:
progress.set_content_size(
response.headers["Content-length"]
)
for chunk in response.iter_content(chunk_size=chunk_size):
stream.write(chunk)
progress.add_transferred_chunk(len(chunk))
content_size = progress.get_content_size()
transferred = progress.get_transferred_size()
if content_size is not None and transferred != content_size:
# Connection was closed before all content was received
raise requests.exceptions.ConnectionError(
f"Downloaded {transferred} out of {content_size}"
f" bytes from '{url}'."
)
break
except (
requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
requests.exceptions.ChunkedEncodingError,
):
if attempt == retries - 1:
raise
progress.next_attempt()
if api_prepended:
self.log.warning(
f"Auto-fixed endpoint '{endpoint}' -> 'api/{endpoint}'."
" Please fix the endpoint passed to the function."
)
[docs] def download_file_to_stream(
self,
endpoint: str,
stream: StreamType,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> TransferProgress:
"""Download file from AYON server to IOStream.
Endpoint can be full url (must start with 'base_url' of api object).
Progress object can be used to track download. Can be used when
download happens in thread and other thread want to catch changes over
time.
Todos:
Use retries and timeout.
Return RestApiResponse.
Args:
endpoint (str): Endpoint or URL to file that should be downloaded.
stream (StreamType): Stream where output will
be stored.
chunk_size (int | None): Size of chunks that are received
in single loop.
progress (TransferProgress | None): Object that gives ability
to track download progress.
"""
if not chunk_size:
chunk_size = self.default_download_chunk_size
if progress is None:
progress = TransferProgress()
progress.set_started()
try:
self._download_file_to_stream(
endpoint, stream, chunk_size, progress
)
except Exception as exc:
progress.set_failed(str(exc))
raise
finally:
progress.set_transfer_done()
return progress
[docs] def download_file(
self,
endpoint: str,
filepath: str,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> TransferProgress:
"""Download file from AYON server.
Endpoint can be full url (must start with 'base_url' of api object).
Progress object can be used to track download. Can be used when
download happens in thread and other thread want to catch changes over
time.
Todos:
Use retries and timeout.
Return RestApiResponse.
Args:
endpoint (str): Endpoint or URL to file that should be downloaded.
filepath (str): Path where file will be downloaded.
chunk_size (int | None): Size of chunks that are received
in single loop.
progress (TransferProgress | None): Object that gives ability
to track download progress.
"""
# Create dummy object so the function does not have to check
# 'progress' variable everywhere
if progress is None:
progress = TransferProgress()
progress.set_destination_url(filepath)
dst_directory = os.path.dirname(filepath)
os.makedirs(dst_directory, exist_ok=True)
try:
with open(filepath, "wb") as stream:
self.download_file_to_stream(
endpoint, stream, chunk_size, progress
)
except Exception as exc:
progress.set_failed(str(exc))
raise
return progress
[docs] def upload_project_file(
self,
project_name: str,
filepath: str,
*,
content_type: str | None = None,
filename: str | None = None,
file_id: str | None = None,
activity_id: str | None = None,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> requests.Response:
"""Upload project file from a filepath.
Project files are usually binary files, such as images, videos,
or other media files that can be accessed via api endpoint
'{server url}/api/projects/{project_name}/files/{file_id}'.
Args:
project_name (str): Project name.
filepath (str): Path where file will be downloaded.
content_type (str | None): MIME type of file.
filename (str | None): Server filename, filename from filepath
is used if not passed.
file_id (str | None): File id.
activity_id (str | None): To which activity is file related.
chunk_size (int | None): Size of chunks that are received
in single loop.
progress (TransferProgress | None): Object that gives ability
to track download progress.
Returns:
requests.Response: Requests response.
"""
if not filename:
filename = os.path.basename(filepath)
if not content_type:
content_type = get_media_mime_type(filepath)
if not content_type:
content_type = "application/octet-stream"
headers = {}
if file_id:
headers["x-file-id"] = file_id
if activity_id:
headers["x-activity-id"] = activity_id
return self.upload_file(
f"api/projects/{project_name}/files",
filepath,
content_type=content_type,
filename=filename,
chunk_size=chunk_size,
progress=progress,
headers=headers,
request_type=RequestTypes.post,
)
[docs] def upload_project_file_from_stream(
self,
project_name: str,
stream: StreamType,
filename: str,
*,
content_type: str | None = None,
file_id: str | None = None,
activity_id: str | None = None,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> requests.Response:
"""Upload project file from a filepath.
Project files are usually binary files, such as images, videos,
or other media files that can be accessed via api endpoint
'{server url}/api/projects/{project_name}/files/{file_id}'.
Args:
project_name (str): Project name.
stream (StreamType): Stream used as source for upload.
filename (str): Name of file on server.
content_type (str | None): MIME type of file.
file_id (str | None): File id.
activity_id (str | None): To which activity is file related.
chunk_size (int | None): Size of chunks that are received
in single loop.
progress (TransferProgress | None): Object that gives ability
to track download progress.
Returns:
requests.Response: Requests response.
"""
if not content_type:
stream.seek(0)
content_type = get_media_mime_type_for_stream(stream)
if not content_type:
content_type = "application/octet-stream"
headers = {}
if file_id:
headers["x-file-id"] = file_id
if activity_id:
headers["x-activity-id"] = activity_id
return self.upload_file_from_stream(
f"api/projects/{project_name}/files",
stream,
content_type=content_type,
filename=filename,
chunk_size=chunk_size,
progress=progress,
headers=headers,
request_type=RequestTypes.post,
)
[docs] def download_project_file(
self,
project_name: str,
file_id: str,
filepath: str,
*,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> TransferProgress:
"""Download project file to filepath.
Project files are usually binary files, such as images, videos,
or other media files that can be accessed via api endpoint
'{server url}/api/projects/{project_name}/files/{file_id}'.
Args:
project_name (str): Project name.
file_id (str): File id.
filepath (str): Path where file will be downloaded.
chunk_size (int | None): Size of chunks that are received
in single loop.
progress (TransferProgress | None): Object that gives ability
to track download progress.
Returns:
TransferProgress: Progress object.
"""
return self.download_file(
f"api/projects/{project_name}/files/{file_id}",
filepath,
chunk_size=chunk_size,
progress=progress,
)
[docs] def download_project_file_to_stream(
self,
project_name: str,
file_id: str,
stream: StreamType,
*,
chunk_size: int | None = None,
progress: TransferProgress | None = None,
) -> TransferProgress:
"""Download project file to a stream.
Project files are usually binary files, such as images, videos,
or other media files that can be accessed via api endpoint
'{server url}/api/projects/{project_name}/files/{file_id}'.
Args:
project_name (str): Project name.
file_id (str): File id.
stream (StreamType): Stream where output will be stored.
chunk_size (int | None): Size of chunks that are received
in single loop.
progress (TransferProgress | None): Object that gives ability
to track download progress.
Returns:
TransferProgress: Progress object.
"""
return self.download_file_to_stream(
f"api/projects/{project_name}/files/{file_id}",
stream,
chunk_size=chunk_size,
progress=progress,
)
[docs] def delete_project_file(self, project_name: str, file_id: str) -> None:
"""Delete project file."""
response = self.delete(f"projects/{project_name}/files/{file_id}")
response.raise_for_status()
@staticmethod
def _upload_chunks_iter(
file_stream: StreamType,
progress: TransferProgress,
chunk_size: int,
) -> Generator[bytes, None, None]:
"""Generator that yields chunks of file.
Args:
file_stream (StreamType): Byte stream.
progress (TransferProgress): Object to track upload progress.
chunk_size (int): Size of chunks that are uploaded at once.
Yields:
bytes: Chunk of file.
"""
file_stream.seek(0)
while True:
chunk = file_stream.read(chunk_size)
if not chunk:
break
progress.add_transferred_chunk(len(chunk))
yield chunk
def _upload_file(
self,
endpoint: str,
stream: StreamType,
progress: TransferProgress,
request_type: RequestType | None = None,
chunk_size: int | None = None,
*,
content_type: str | None = None,
filename: str | None = None,
**kwargs
) -> requests.Response:
"""Upload file to server.
Args:
endpoint (str): Endpoint used to upload.
stream (StreamType): File stream.
progress (TransferProgress): Object that gives ability to track
progress.
request_type (RequestType | None): Type of request that will
be used. Default is PUT.
chunk_size (int | None): Size of chunks that are uploaded
at once.
**kwargs (Any): Additional arguments that will be passed
to request function.
Returns:
requests.Response: Server response.
"""
if request_type is None:
request_type = RequestTypes.put
endpoint = endpoint.lstrip("/")
url = self._endpoint_to_url(endpoint, use_rest=False)
progress.set_destination_url(url)
headers = kwargs.get("headers")
if headers is None:
kwargs["headers"] = headers = {}
headers_keys_by_low_key = {key.lower(): key for key in headers}
if self._session is None:
for key, value in self.get_headers().items():
orig_key = headers_keys_by_low_key.get(key)
if not orig_key:
headers[key] = value
post_func = self._base_functions_mapping[request_type]
else:
post_func = self._session_functions_mapping[request_type]
if not chunk_size:
chunk_size = self.default_upload_chunk_size
for key, value in (
("x-file-name", filename),
("Content-Type", content_type),
):
if not value:
continue
orig_key = headers_keys_by_low_key.get(key.lower())
if orig_key:
headers.pop(orig_key)
headers[key] = value
retries = max(self.max_retries, 1)
response = None
# Get size of file
stream.seek(0, io.SEEK_END)
size = stream.tell()
# Set content size to progress object
progress.set_content_size(size)
api_prepended = False
for attempt in range(retries):
try:
response = post_func(
url,
data=self._upload_chunks_iter(
stream, progress, chunk_size
),
**kwargs
)
# Auto-fix missing 'api/'
if response.status_code in (404, 405) and not api_prepended:
api_prepended = True
if (
not endpoint.startswith(self._base_url)
and not endpoint.startswith("api/")
):
url = self._endpoint_to_url(endpoint, use_rest=True)
progress.set_destination_url(url)
continue
break
except (
requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
):
if attempt == retries - 1:
raise
progress.next_attempt()
progress.reset_transferred()
response.raise_for_status()
if api_prepended:
self.log.warning(
f"Auto-fixed endpoint '{endpoint}' -> 'api/{endpoint}'."
" Please fix the endpoint passed to the function."
)
return response
[docs] def upload_file_from_stream(
self,
endpoint: str,
stream: StreamType,
progress: TransferProgress | None = None,
request_type: RequestType | None = None,
*,
content_type: str | None = None,
filename: str | None = None,
**kwargs
) -> requests.Response:
"""Upload file to server from bytes.
Todos:
Use retries and timeout.
Return RestApiResponse.
Args:
endpoint (str): Endpoint or url where file will be uploaded.
stream (StreamType): File content stream.
progress (TransferProgress | None): Object that gives ability
to track upload progress.
request_type (RequestType | None): Type of request that will
be used to upload file.
content_type (str | None): MIME type of the file.
filename (str | None): Filename of file on server.
**kwargs (Any): Additional arguments that will be passed
to request function.
Returns:
requests.Response: Response object
"""
# Create dummy object so the function does not have to check
# 'progress' variable everywhere
if progress is None:
progress = TransferProgress()
progress.set_started()
try:
return self._upload_file(
endpoint,
stream,
progress,
request_type,
content_type=content_type,
filename=filename,
**kwargs
)
except Exception as exc:
progress.set_failed(str(exc))
raise
finally:
progress.set_transfer_done()
[docs] def upload_file(
self,
endpoint: str,
filepath: str,
progress: TransferProgress | None = None,
request_type: RequestType | None = None,
*,
content_type: str | None = None,
filename: str | None = None,
**kwargs
) -> requests.Response:
"""Upload file to server.
Todos:
Use retries and timeout.
Return RestApiResponse.
Args:
endpoint (str): Endpoint or url where file will be uploaded.
filepath (str): Source filepath.
progress (TransferProgress | None): Object that gives ability
to track upload progress.
request_type (RequestType | None): Type of request that will
be used to upload file.
content_type (str | None): MIME type of the file.
filename (str | None): Filename of file on server.
**kwargs (Any): Additional arguments that will be passed
to request function.
Returns:
requests.Response: Response object
"""
if progress is None:
progress = TransferProgress()
progress.set_source_url(filepath)
with open(filepath, "rb") as stream:
return self.upload_file_from_stream(
endpoint,
stream,
progress,
request_type,
content_type=content_type,
filename=filename,
**kwargs
)
[docs] def upload_reviewable(
self,
project_name: str,
version_id: str,
filepath: str,
label: str | None = None,
content_type: str | None = None,
filename: str | None = None,
progress: TransferProgress | None = None,
**kwargs
) -> requests.Response:
"""Upload reviewable file to server.
Args:
project_name (str): Project name.
version_id (str): Version id.
filepath (str): Reviewable file path to upload.
label (str | None): Reviewable label. Filled automatically
server side with filename.
content_type (str | None): MIME type of the file.
filename (str | None): User as original filename. Filename from
'filepath' is used when not filled.
progress (TransferProgress | None): Progress.
Returns:
requests.Response: Server response.
"""
if not content_type:
content_type = get_media_mime_type(filepath)
if not content_type:
raise ValueError(
f"Could not determine MIME type of file '{filepath}'"
)
# Fill original filename if not explicitly defined
if not filename:
filename = os.path.basename(filepath)
query = prepare_query_string({"label": label or None})
endpoint = (
f"api/projects/{project_name}"
f"/versions/{version_id}/reviewables{query}"
)
return self.upload_file(
endpoint,
filepath,
progress=progress,
content_type=content_type,
filename=filename,
request_type=RequestTypes.post,
**kwargs
)
[docs] def trigger_server_restart(self) -> None:
"""Trigger server restart.
Restart may be required when a change of specific value happened on
server.
"""
result = self.post("system/restart")
if result.status_code != 204:
# TODO add better exception
raise ValueError("Failed to restart server")
[docs] def query_graphql(
self,
query: str,
variables: dict[str, Any] | None = None,
) -> GraphQlResponse:
"""Execute GraphQl query.
Args:
query (str): GraphQl query string.
variables (dict[str, Any] | None): Variables that can be
used in query.
Returns:
GraphQlResponse: Response from server.
"""
data = {"query": query, "variables": variables or {}}
response = self._do_rest_request(
RequestTypes.post,
self._graphql_url,
json=data
)
response.raise_for_status()
return GraphQlResponse(response)
[docs] def get_graphql_schema(self) -> dict[str, Any]:
return self.query_graphql(INTROSPECTION_QUERY).data["data"]
[docs] def get_server_schema(self) -> dict[str, Any] | None:
"""Get server schema with info, url paths, components etc.
Todos:
Cache schema - How to find out it is outdated?
Returns:
dict[str, Any]: Full server schema.
"""
url = f"{self._base_url}/openapi.json"
response = self._do_rest_request(RequestTypes.get, url)
if response:
return response.data
return None
[docs] def get_schemas(self) -> dict[str, Any]:
"""Get components schema.
Name of components does not match entity type names e.g. 'project' is
under 'ProjectModel'. We should find out some mapping. Also, there
are properties which don't have information about reference to object
e.g. 'config' has just object definition without reference schema.
Returns:
dict[str, Any]: Component schemas.
"""
server_schema = self.get_server_schema()
return server_schema["components"]["schemas"]
[docs] def get_default_fields_for_type(self, entity_type: str) -> set[str]:
"""Default fields for entity type.
Returns most of commonly used fields from server.
Args:
entity_type (str): Name of entity type.
Returns:
set[str]: Fields that should be queried from server.
"""
# Event does not have attributes
if entity_type == "event":
return set(DEFAULT_EVENT_FIELDS)
if entity_type == "activity":
return set(DEFAULT_ACTIVITY_FIELDS)
if entity_type == "productType":
return set(DEFAULT_PRODUCT_TYPE_FIELDS)
if entity_type == "project":
entity_type_defaults = set(DEFAULT_PROJECT_FIELDS)
maj_v, min_v, patch_v, _, _ = self.server_version_tuple
if (maj_v, min_v, patch_v) > (1, 10, 0):
entity_type_defaults.add("productTypes")
elif entity_type == "folder":
entity_type_defaults = set(DEFAULT_FOLDER_FIELDS)
elif entity_type == "task":
entity_type_defaults = set(DEFAULT_TASK_FIELDS)
elif entity_type == "product":
entity_type_defaults = set(DEFAULT_PRODUCT_FIELDS)
maj_v, min_v, patch_v, _, _ = self.server_version_tuple
if self.is_product_base_type_supported():
entity_type_defaults.add("productBaseType")
elif entity_type == "version":
entity_type_defaults = set(DEFAULT_VERSION_FIELDS)
elif entity_type == "representation":
entity_type_defaults = (
DEFAULT_REPRESENTATION_FIELDS
| REPRESENTATION_FILES_FIELDS
)
if not self.graphql_allows_traits_in_representations:
entity_type_defaults.discard("traits")
elif entity_type == "workfile":
entity_type_defaults = set(DEFAULT_WORKFILE_INFO_FIELDS)
elif entity_type == "user":
entity_type_defaults = set(DEFAULT_USER_FIELDS)
elif entity_type == "entityList":
entity_type_defaults = set(DEFAULT_ENTITY_LIST_FIELDS)
else:
raise ValueError(f"Unknown entity type \"{entity_type}\"")
entity_type_defaults.add("allAttrib")
return entity_type_defaults
[docs] def get_rest_entity_by_id(
self,
project_name: str,
entity_type: str,
entity_id: str,
) -> AnyEntityDict | None:
"""Get entity using REST on a project by its id.
Args:
project_name (str): Name of project where entity is.
entity_type (Literal["folder", "task", "product", "version"]): The
entity type which should be received.
entity_id (str): Id of entity.
Returns:
AnyEntityDict | None: Received entity data.
"""
if not all((project_name, entity_type, entity_id)):
return None
response = self.get(
f"projects/{project_name}/{entity_type}s/{entity_id}"
)
if response.status == 200:
return response.data
return None
# --- Batch operations processing ---
[docs] def send_batch_operations(
self,
project_name: str,
operations: list[dict[str, Any]],
*,
can_fail: bool = False,
wait_for_events: bool = False,
raise_on_fail: bool = True,
) -> list[dict[str, Any]]:
"""Post multiple CRUD operations to server.
When multiple changes should be made on server side this is the best
way to go. It is possible to pass multiple operations to process on a
server side and do the changes in a transaction.
Args:
project_name (str): On which project should be operations
processed.
operations (list[dict[str, Any]]): Operations to be processed.
can_fail (bool): Server will try to process all
operations even if one of them fails.
wait_for_events (bool): Wait for events to be processed on server.
raise_on_fail (bool): Raise exception if an operation
fails. You can handle failed operations on your own
when set to 'False'.
Raises:
ValueError: Operations can't be converted to json string.
FailedOperations: When output does not contain server operations
or 'raise_on_fail' is enabled and any operation fails.
Returns:
list[dict[str, Any]]: Operations result with process details.
"""
return self._send_batch_operations(
f"projects/{project_name}/operations",
operations,
can_fail=can_fail,
wait_for_events=wait_for_events,
raise_on_fail=raise_on_fail,
)
[docs] def send_background_batch_operations(
self,
project_name: str,
operations: list[dict[str, Any]],
*,
can_fail: bool = False,
wait_for_events: bool = False,
raise_on_fail: bool = True,
wait: bool = False,
) -> BackgroundOperationTask:
"""Post multiple CRUD operations to server.
When multiple changes should be made on server side this is the best
way to go. It is possible to pass multiple operations to process on a
server side and do the changes in a transaction.
Compared to 'send_batch_operations' this function creates a task on
server which then can be periodically checked for a status and
receive it's result.
When used with 'wait' set to 'True' this method blocks until task is
finished. Which makes it work as 'send_batch_operations'
but safer for large operations batch as is not bound to
response timeout.
Args:
project_name (str): On which project should be operations
processed.
operations (list[dict[str, Any]]): Operations to be processed.
can_fail (bool): Server will try to process all
operations even if one of them fails.
wait_for_events (bool): Wait for events to be processed on server.
raise_on_fail (bool): Raise exception if an operation
fails. You can handle failed operations on your own
when set to 'False'. Used when 'wait' is enabled.
wait (bool): Wait for operations to end.
Raises:
ValueError: Operations can't be converted to json string.
FailedOperations: When output does not contain server operations
or 'raise_on_fail' is enabled and any operation fails.
Returns:
BackgroundOperationTask: Background operation.
"""
operations_body = self._prepare_operations_body(operations)
response = self.post(
f"projects/{project_name}/operations/background",
operations=operations_body,
canFail=can_fail,
waitForEvents=wait_for_events,
)
response.raise_for_status()
if not wait:
return response.data
task_id = response["id"]
time.sleep(0.1)
while True:
op_status = self.get_background_operations_status(
project_name, task_id
)
if op_status["status"] == "completed":
break
time.sleep(1)
if raise_on_fail:
self._validate_operations_result(
op_status["result"], operations_body
)
return op_status
[docs] def get_background_operations_status(
self, project_name: str, task_id: str
) -> BackgroundOperationTask:
"""Get status of background operations task.
Args:
project_name (str): Project name.
task_id (str): Backgorund operation task id.
Returns:
BackgroundOperationTask: Background operation.
"""
response = self.get(
f"projects/{project_name}/operations/background/{task_id}"
)
response.raise_for_status()
return response.data
def _prepare_operations_body(
self, operations: list[dict[str, Any]]
) -> list[dict[str, Any]]:
operations_body = []
for operation in operations:
if not operation:
continue
op_id = operation.get("id")
if not op_id:
op_id = create_entity_id()
operation["id"] = op_id
try:
body = json.loads(
json.dumps(operation, default=entity_data_json_default)
)
except (TypeError, ValueError):
raise ValueError("Couldn't json parse body: {}".format(
json.dumps(
operation, indent=4, default=failed_json_default
)
))
operations_body.append(body)
return operations_body
def _send_batch_operations(
self,
uri: str,
operations: list[dict[str, Any]],
can_fail: bool,
wait_for_events: bool,
raise_on_fail: bool,
) -> list[dict[str, Any]]:
if not operations:
return []
operations_body = self._prepare_operations_body(operations)
if not operations_body:
return []
response = self.post(
uri,
operations=operations_body,
canFail=can_fail,
waitForEvents=wait_for_events,
)
op_results = response.get("operations")
if op_results is None:
detail = response.get("detail")
if detail:
raise FailedOperations(f"Operation failed. Detail: {detail}")
raise FailedOperations(
f"Operation failed. Content: {response.text}"
)
if raise_on_fail:
self._validate_operations_result(response.data, operations_body)
return op_results
def _validate_operations_result(
self,
result: dict[str, Any],
operations_body: list[dict[str, Any]],
) -> None:
if result.get("success"):
return None
self.log.warning(
"Operations failed. Server response:\n%s",
json.dumps(result, indent=4, default=str),
)
for op_result in result.get("operations") or []:
if op_result["success"]:
continue
operation_id = op_result["id"]
operation = next(
(op for op in operations_body if op["id"] == operation_id),
op_result,
)
detail = op_result["detail"]
raise FailedOperations(
f"Operation \"{operation_id}\" failed with data:"
f"\n{json.dumps(operation, indent=4, default=str)}"
f"\nDetail: {detail}."
)
# Server did not report which operation failed
raise FailedOperations(
"Operations failed. Server response:"
f"\n{json.dumps(result, indent=4, default=str)}"
)
def _prepare_fields(
self,
entity_type: str,
fields: set[str],
own_attributes: bool = False,
):
if not fields:
return
add_all_attrib = False
for field in tuple(fields):
if field == "attrib" or field.startswith("attrib."):
fields.discard(field)
add_all_attrib = True
if own_attributes:
if entity_type == "project":
add_all_attrib = True
elif entity_type in {"folder", "task"}:
# Attribute values are needed to fill 'ownAttrib'
add_all_attrib = True
fields.add("ownAttrib")
if add_all_attrib:
fields.add("allAttrib")
if entity_type != "project":
return
# Use 'data' to fill 'bundle' data
if "bundle" in fields:
fields.remove("bundle")
fields.add("data")
maj_v, min_v, patch_v, _, _ = self.server_version_tuple
if "folderTypes" in fields:
fields.remove("folderTypes")
folder_types_fields = set(DEFAULT_FOLDER_TYPE_FIELDS)
if (maj_v, min_v, patch_v) > (1, 10, 0):
folder_types_fields |= {"shortName"}
fields |= {f"folderTypes.{name}" for name in folder_types_fields}
if "taskTypes" in fields:
fields.remove("taskTypes")
task_types_fields = set(DEFAULT_TASK_TYPE_FIELDS)
if (maj_v, min_v, patch_v) > (1, 10, 0):
task_types_fields |= {"color", "icon", "shortName"}
fields |= {f"taskTypes.{name}" for name in task_types_fields}
for field, default_fields in (
("statuses", DEFAULT_PROJECT_STATUSES_FIELDS),
("tags", DEFAULT_PROJECT_TAGS_FIELDS),
("linkTypes", DEFAULT_PROJECT_TAGS_FIELDS),
):
if (maj_v, min_v, patch_v) <= (1, 10, 0):
break
if field in fields:
fields.remove(field)
fields |= {f"{field}.{name}" for name in default_fields}
if "productTypes" in fields:
fields.remove("productTypes")
fields |= {
f"productTypes.{name}"
for name in self.get_default_fields_for_type(
"productType"
)
}
def _prepare_link_fields(self, fields: set[str]) -> None:
if "links" not in fields:
return
fields.discard("links")
for field in DEFAULT_LINK_FIELDS:
fields.add(f"links.{field}")
if self.links_graphql_support_data():
fields.add("links.data")
def _prepare_advanced_filters(
self, filters: str | dict[str, Any] | None
) -> str | None:
if not filters:
return None
if isinstance(filters, dict):
return json.dumps(filters)
return filters
def _convert_entity_data(self, entity: AnyEntityDict) -> None:
if not entity:
return
if "data" in entity:
entity_data = entity["data"] or {}
if isinstance(entity_data, str):
entity_data = json.loads(entity_data)
entity["data"] = entity_data
all_attrib = entity.get("allAttrib")
if isinstance(all_attrib, str):
# NOTE: This expects server returns all attributes available for
# the entity type.
entity["attrib"] = json.loads(all_attrib)