Activities endpoints

automated_api.py

@@ -114,25 +114,40 @@ def prepare_docstring(func):
     return f'"""{docstring}{line_char}\n"""'
 
 
-def _get_typehint(param, api_globals):
-    if param.annotation is inspect.Parameter.empty:
-        return None
-
-    an = param.annotation
-    if inspect.isclass(an):
-        return an.__name__
+def _get_typehint(annotation, api_globals):
+    if inspect.isclass(annotation):
+        return annotation.__name__
+
+    typehint = (
+        str(annotation)
+        .replace("typing.", "")
+        .replace("NoneType", "None")
+    )
+    forwardref_regex = re.compile(
+        r"(?P<full>ForwardRef\('(?P<name>[a-zA-Z0-9]+)'\))"
+    )
+    for item in forwardref_regex.finditer(str(typehint)):
+        groups = item.groupdict()
+        name = groups["name"]
+        typehint = typehint.replace(groups["full"], f'"{name}"')
 
-    typehint = str(an).replace("typing.", "")
     try:
         # Test if typehint is valid for known '_api' content
         exec(f"_: {typehint} = None", api_globals)
     except NameError:
+        print("Unknown typehint:", typehint)
         typehint = f'"{typehint}"'
     return typehint
 
 
+def _get_param_typehint(param, api_globals):
+    if param.annotation is inspect.Parameter.empty:
+        return None
+    return _get_typehint(param.annotation, api_globals)
+
+
 def _add_typehint(param_name, param, api_globals):
-    typehint = _get_typehint(param, api_globals)
+    typehint = _get_param_typehint(param, api_globals)
     if not typehint:
         return param_name
     return f"{param_name}: {typehint}"
@@ -154,7 +169,7 @@ def _kw_default_to_str(param_name, param, api_globals):
         raise TypeError("Unknown default value type")
     else:
         default = repr(default)
-    typehint = _get_typehint(param, api_globals)
+    typehint = _get_param_typehint(param, api_globals)
     if typehint:
         return f"{param_name}: {typehint} = {default}"
     return f"{param_name}={default}"
@@ -216,7 +231,8 @@ def sig_params_to_str(sig, param_names, api_globals, indent=0):
         func_params_str = f"(\n{lines_str}\n{base_indent_str})"
 
     if sig.return_annotation is not inspect.Signature.empty:
-        func_params_str += f" -> {sig.return_annotation}"
+        return_typehint = _get_typehint(sig.return_annotation, api_globals)
+        func_params_str += f" -> {return_typehint}"
 
     body_params_str = "()"
     if body_params:

ayon_api/__init__.py

@@ -72,6 +72,11 @@
     dispatch_event,
     delete_event,
     enroll_event_job,
+    get_activities,
+    get_activity_by_id,
+    create_activity,
+    update_activity,
+    delete_activity,
     download_file_to_stream,
     download_file,
     upload_file_from_stream,
@@ -234,6 +239,7 @@
     get_representations_links,
     get_representation_links,
     send_batch_operations,
+    send_activities_batch_operations,
 )
 
 
@@ -309,6 +315,11 @@
     "dispatch_event",
     "delete_event",
     "enroll_event_job",
+    "get_activities",
+    "get_activity_by_id",
+    "create_activity",
+    "update_activity",
+    "delete_activity",
     "download_file_to_stream",
     "download_file",
     "upload_file_from_stream",
@@ -471,4 +482,5 @@
     "get_representations_links",
     "get_representation_links",
     "send_batch_operations",
+    "send_activities_batch_operations",
 )

ayon_api/_api.py

@@ -11,7 +11,8 @@
 
 import os
 import socket
-from typing import Optional
+import typing
+from typing import Optional, List, Dict, Iterable, Generator, Any
 
 from .constants import (
     SERVER_URL_ENV_KEY,
@@ -24,6 +25,9 @@
     get_default_settings_variant as _get_default_settings_variant,
 )
 
+if typing.TYPE_CHECKING:
+    from ._typing import ActivityType, ActivityReferenceType
+
 
 class GlobalServerAPI(ServerAPI):
     """Extended server api which also handles storing tokens and url.
@@ -1121,6 +1125,176 @@ def enroll_event_job(
     )
 
 
+def get_activities(
+    project_name: str,
+    activity_ids: Optional[Iterable[str]] = None,
+    activity_types: Optional[Iterable["ActivityType"]] = None,
+    entity_ids: Optional[Iterable[str]] = None,
+    entity_names: Optional[Iterable[str]] = None,
+    entity_type: Optional[str] = None,
+    changed_after: Optional[str] = None,
+    changed_before: Optional[str] = None,
+    reference_types: Optional[Iterable["ActivityReferenceType"]] = None,
+    fields: Optional[Iterable[str]] = None,
+) -> Generator[Dict[str, Any], None, None]:
+    """Get activities from server with filtering options.
+
+    Args:
+        project_name (str): Project on which activities happened.
+        activity_ids (Optional[Iterable[str]]): Activity ids.
+        activity_types (Optional[Iterable[ActivityType]]): Activity types.
+        entity_ids (Optional[Iterable[str]]): Entity ids.
+        entity_names (Optional[Iterable[str]]): Entity names.
+        entity_type (Optional[str]): Entity type.
+        changed_after (Optional[str]): Return only activities changed
+            after given iso datetime string.
+        changed_before (Optional[str]): Return only activities changed
+            before given iso datetime string.
+        reference_types (Optional[Iterable[ActivityReferenceType]]):
+            Reference types filter. Defaults to `['origin']`.
+        fields (Optional[Iterable[str]]): Fields that should be received
+            for each activity.
+
+    Returns:
+        Generator[dict[str, Any]]: Available activities matching filters.
+
+    """
+    con = get_server_api_connection()
+    return con.get_activities(
+        project_name=project_name,
+        activity_ids=activity_ids,
+        activity_types=activity_types,
+        entity_ids=entity_ids,
+        entity_names=entity_names,
+        entity_type=entity_type,
+        changed_after=changed_after,
+        changed_before=changed_before,
+        reference_types=reference_types,
+        fields=fields,
+    )
+
+
+def get_activity_by_id(
+    project_name: str,
+    activity_id: str,
+    reference_types: Optional[Iterable["ActivityReferenceType"]] = None,
+    fields: Optional[Iterable[str]] = None,
+) -> Optional[Dict[str, Any]]:
+    """Get activity by id.
+
+    Args:
+        project_name (str): Project on which activity happened.
+        activity_id (str): Activity id.
+        fields (Optional[Iterable[str]]): Fields that should be received
+            for each activity.
+
+    Returns:
+        Optional[Dict[str, Any]]: Activity data or None if activity is not
+            found.
+
+    """
+    con = get_server_api_connection()
+    return con.get_activity_by_id(
+        project_name=project_name,
+        activity_id=activity_id,
+        reference_types=reference_types,
+        fields=fields,
+    )
+
+
+def create_activity(
+    project_name: str,
+    entity_id: str,
+    entity_type: str,
+    activity_type: "ActivityType",
+    activity_id: Optional[str] = None,
+    body: Optional[str] = None,
+    file_ids: Optional[List[str]] = None,
+    timestamp: Optional[str] = None,
+    data: Optional[Dict[str, Any]] = None,
+) -> str:
+    """Create activity on a project.
+
+    Args:
+        project_name (str): Project on which activity happened.
+        entity_id (str): Entity id.
+        entity_type (str): Entity type.
+        activity_type (ActivityType): Activity type.
+        activity_id (Optional[str]): Activity id.
+        body (Optional[str]): Activity body.
+        file_ids (Optional[List[str]]): List of file ids attached
+            to activity.
+        timestamp (Optional[str]): Activity timestamp.
+        data (Optional[Dict[str, Any]]): Additional data.
+
+    Returns:
+        str: Activity id.
+
+    """
+    con = get_server_api_connection()
+    return con.create_activity(
+        project_name=project_name,
+        entity_id=entity_id,
+        entity_type=entity_type,
+        activity_type=activity_type,
+        activity_id=activity_id,
+        body=body,
+        file_ids=file_ids,
+        timestamp=timestamp,
+        data=data,
+    )
+
+
+def update_activity(
+    project_name: str,
+    activity_id: str,
+    body: Optional[str] = None,
+    file_ids: Optional[List[str]] = None,
+    append_file_ids: Optional[bool] = False,
+    data: Optional[Dict[str, Any]] = None,
+):
+    """Update activity by id.
+
+    Args:
+        project_name (str): Project on which activity happened.
+        activity_id (str): Activity id.
+        body (str): Activity body.
+        file_ids (Optional[List[str]]): List of file ids attached
+            to activity.
+        append_file_ids (Optional[bool]): Append file ids to existing
+            list of file ids.
+        data (Optional[Dict[str, Any]]): Update data in activity.
+
+    """
+    con = get_server_api_connection()
+    return con.update_activity(
+        project_name=project_name,
+        activity_id=activity_id,
+        body=body,
+        file_ids=file_ids,
+        append_file_ids=append_file_ids,
+        data=data,
+    )
+
+
+def delete_activity(
+    project_name: str,
+    activity_id: str,
+):
+    """Delete activity by id.
+
+    Args:
+        project_name (str): Project on which activity happened.
+        activity_id (str): Activity id to remove.
+
+    """
+    con = get_server_api_connection()
+    return con.delete_activity(
+        project_name=project_name,
+        activity_id=activity_id,
+    )
+
+
 def download_file_to_stream(
     endpoint,
     stream,
@@ -6396,3 +6570,43 @@ def send_batch_operations(
         can_fail=can_fail,
         raise_on_fail=raise_on_fail,
     )
+
+
+def send_activities_batch_operations(
+    project_name,
+    operations,
+    can_fail=False,
+    raise_on_fail=True,
+):
+    """Post multiple CRUD activities 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 (Optional[bool]): Server will try to process all
+            operations even if one of them fails.
+        raise_on_fail (Optional[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.
+
+    """
+    con = get_server_api_connection()
+    return con.send_activities_batch_operations(
+        project_name=project_name,
+        operations=operations,
+        can_fail=can_fail,
+        raise_on_fail=raise_on_fail,
+    )

ayon_api/_typing.py

@@ -0,0 +1,19 @@
+from typing import Literal
+
+ActivityType = Literal[
+    "comment",
+    "watch",
+    "reviewable",
+    "status.change",
+    "assignee.add",
+    "assignee.remove",
+    "version.publish"
+]
+
+ActivityReferenceType = Literal[
+    "origin",
+    "mention",
+    "author",
+    "relation",
+    "watching",
+]

ayon_api/constants.py

@@ -185,3 +185,13 @@
     "description",
     "author",
 }
+
+DEFAULT_ACTIVITY_FIELDS = {
+    "activityId",
+    "activityType",
+    "activityData",
+    "body",
+    "entityId",
+    "entityType",
+    "author.name",
+}