"""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 time

import logging

import platform

import uuid

from contextlib import contextmanager

import typing

from typing import (

Iterable, Generator, Any, Literal, ContextManager

)

import requests

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 .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,

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,

)

if typing.TYPE_CHECKING:

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\-.]*))?"

)

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()

@dataclass

class TokenInfo:

token: str | None = None

is_valid: bool | None = None

is_service: bool | None = None

unauthorized_response: requests.Response | None = None

class ServerAPI(

InstallersAPI,

DependencyPackagesAPI,

SecretsAPI,

BundlesAddonsAPI,

EventsAPI,

AttributesAPI,

ProjectsAPI,

FoldersAPI,

TasksAPI,

ProductsAPI,

VersionsAPI,

RepresentationsAPI,

WorkfilesAPI,

ThumbnailsAPI,

ActivitiesAPI,

ActionsAPI,

LinksAPI,

ListsAPI,

):

"""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

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()

@property

def log(self) -> logging.Logger:

return self._log

def get_base_url(self) -> str:

return self._base_url

def get_rest_url(self) -> str:

return self._rest_url

base_url = property(get_base_url)

rest_url = property(get_rest_url)

def get_ssl_verify(self) -> bool | str | None:

"""Enable ssl verification.

Returns:

bool | str | None: Current state of ssl verification.

"""

return self._ssl_verify

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

if self._session is not None:

self._session.verify = ssl_verify

def get_cert(self) -> str | None:

"""Current cert file used for connection to server.

Returns:

str | None: Path to cert file.

"""

return self._cert

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

if self._session is not None:

self._session.cert = cert

ssl_verify = property(get_ssl_verify, set_ssl_verify)

cert = property(get_cert, set_cert)

@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()

@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

def get_timeout(self) -> float:

"""Current value for requests timeout.

Returns:

float: Timeout value in seconds.

"""

return self._timeout

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)

def get_max_retries(self) -> int:

"""Current value for requests max retries.

Returns:

int: Max retries value.

"""

return self._max_retries

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

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)

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

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)

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

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)

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

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

)

def get_sender(self) -> str | None:

"""Sender used to send requests.

Returns:

str | None: Sender name or None.

"""

return self._sender

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)

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

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)

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()

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()

@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:

response = requests.get(

self._base_url,

cert=self._cert,

verify=self._ssl_verify,

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

def validate_server_availability(self) -> None:

if not self.is_server_available:

raise ServerNotReached(

f"Server \"{self._base_url}\" can't be reached"

)

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

# TODO add other possible validations

# - existence of 'user' key in info

# - validate that 'site_id' is in 'sites' in info

# Check server url

self._get_server_info()

user_info = get_user_info_by_token(

self.base_url,

self._token_info.token,

verify=self._ssl_verify,

cert=self._cert,

timeout=self.timeout,

)

self._token_info.is_valid = user_info.is_valid

self._token_info.unauthorized_response = user_info.response

is_service = None

if user_info.is_valid:

is_service = user_info.is_service

self._token_info.is_service = is_service

return self._token_info.is_valid

def set_token(self, token: str | None) -> None:

self.reset_token()

self._token_info.token = token

self.validate_token()

def reset_token(self) -> None:

self._token_info.token = None

self._token_info.is_service = None

self._token_info.is_valid = None

self._token_info.unauthorized_response = None

self.close_session()

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

self.validate_token()

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

def close_session(self) -> None:

if self._session is None:

return

session = self._session

self._session = None

self._session_functions_mapping = {}

session.close()

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)

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

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

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

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

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

def get_users(

self,

project_name: str | None = None,

usernames: Iterable[str] | None = None,

emails: Iterable[str] | None = None,

fields: Iterable[str] | None = None,

) -> 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.

Returns:

Generator[dict[str, Any], None, None]: Queried users.