[pull] master from atlassian-api:master

atlassian/__init__.py

@@ -6,12 +6,13 @@
 from .bitbucket import Bitbucket
 from .bitbucket import Bitbucket as Stash
 from .cloud_admin import CloudAdmin, CloudAdminOrgs, CloudAdminUsers
+from .compass import Compass
 from .confluence import Confluence, ConfluenceBase, ConfluenceCloud, ConfluenceServer
 from .confluence.cloud.cloud import ConfluenceCloud as ConfluenceV2
 from .crowd import Crowd
 from .insight import Insight
 from .insight import Insight as Assets  # used for Insight on-premise
-from .assets import AssetsCloud  # used for Insight Cloud
+from .assets import AssetsCloud, AssetsDataCenter, AssetsServer  # Assets Cloud and Data Center
 from .jira import Jira, JiraCloud, JiraServer, JiraServiceManagement, JiraSoftware, create_jira_cloud
 from .marketplace import MarketPlace
 from .portfolio import Portfolio
@@ -46,6 +47,7 @@ def create_confluence(url, *args, api_version=1, **kwargs):
     "CloudAdminOrgs",
     "CloudAdminUsers",
     "CloudAdmin",
+    "Compass",
     "Portfolio",
     "Bamboo",
     "Stash",
@@ -57,6 +59,8 @@ def create_confluence(url, *args, api_version=1, **kwargs):
     "Insight",
     "Assets",
     "AssetsCloud",
+    "AssetsDataCenter",
+    "AssetsServer",
     "TempoCloud",
     "TempoServer",
     "YogiJiraCloud",

atlassian/assets/__init__.py

@@ -0,0 +1,6 @@
+"""Assets clients for Jira Cloud and Jira Data Center."""
+
+from .assets_cloud import AssetsCloud
+from .assets_server import AssetsDataCenter, AssetsServer
+
+__all__ = ["AssetsCloud", "AssetsDataCenter", "AssetsServer"]

atlassian/assets/assets_server.py

@@ -0,0 +1,24 @@
+"""Jira Service Management Assets Server/Data Center client."""
+
+from .assets_cloud import AssetsCloud
+
+
+class AssetsServer(AssetsCloud):
+    """Assets REST client for Jira Server/Data Center installations.
+
+    Assets Data Center 10.x exposes the REST resources below
+    ``/rest/assets/1.0``.  The implementation shares the resource methods
+    with :class:`~atlassian.assets.assets_cloud.AssetsCloud`; only workspace
+    discovery and the Cloud gateway are skipped.
+    """
+
+    def __init__(self, *args, **kwargs):
+        # Explicitly force the server path even if a caller reuses a config
+        # dictionary that contains ``cloud=True``.
+        kwargs["cloud"] = False
+        super().__init__(*args, **kwargs)
+
+
+# Data Center is the current product name; retain this descriptive alias for
+# callers that used the previous class name.
+AssetsDataCenter = AssetsServer

atlassian/bitbucket/__init__.py

@@ -124,6 +124,17 @@ def check_reindexing_status(self):
         url = self.resource_url("status", api_root="rest/indexing", api_version="latest")
         return self.get(url)
 
+    def get_cluster_info(self):
+        """Get system information for the Bitbucket Data Center cluster.
+
+        This administrative endpoint is available on Bitbucket Server/Data
+        Center only and requires the appropriate administrator permission.
+        """
+        if self.cloud:
+            raise NotImplementedError("Cluster information is only available on Bitbucket Server/Data Center")
+        url = self.resource_url("admin/cluster", api_version="latest")
+        return self.get(url)
+
     def get_users(self, user_filter=None, limit=25, start=0):
         """
         Get list of bitbucket users.
@@ -366,6 +377,50 @@ def delete_repo_hook_script(self, project_key: str, repository_slug: str, script
         url = f"{self._url_repo(project_key, repository_slug, api_version='latest')}/hook-scripts/{script_id}"
         return self.delete(url)
 
+    def get_hook_script(self, script_id: int):
+        """Return the registered global hook script with ``script_id``.
+
+        The result only contains the script metadata; use
+        ``get_hook_script_content()`` to fetch the script body.
+        """
+        return self.get(f"{self._url_hook_scripts()}/{script_id}")
+
+    def get_hook_script_content(self, script_id: int):
+        """Return the raw body of the registered global hook script.
+
+        ``not_json_response`` is used because the endpoint returns the
+        executable script as plain text rather than a JSON payload.
+        """
+        return self.get(
+            f"{self._url_hook_scripts()}/{script_id}/content",
+            not_json_response=True,
+        )
+
+    def update_hook_script(
+        self, script_id: int, content: bytes, name: str, hook_type: str, description: Optional[str] = None
+    ):
+        """Update a registered global hook script on Bitbucket Data Center.
+
+        ``hook_type`` must be ``PRE`` or ``POST``. Like ``create_hook_script``,
+        the API expects a multipart form; ``content`` is the executable script
+        bytes.
+        """
+        if hook_type not in ("PRE", "POST"):
+            raise ValueError("hook_type must be 'PRE' or 'POST'")
+
+        files: Dict[str, Any] = {
+            "content": ("hook-script", content, "application/octet-stream"),
+            "name": (None, name),
+            "type": (None, hook_type),
+        }
+        if description is not None:
+            files["description"] = (None, description)
+        return self.put(f"{self._url_hook_scripts()}/{script_id}", files=files, headers=self.no_check_headers)
+
+    def delete_hook_script(self, script_id: int):
+        """Delete the registered global hook script with ``script_id``."""
+        return self.delete(f"{self._url_hook_scripts()}/{script_id}")
+
     def get_categories(self, project_key, repository_slug=None):
         """
         Get a list of categories assigned to a project or repository.
@@ -981,6 +1036,40 @@ def update_repo(self, project_key, repository_slug, **params):
         url = self._url_repo(project_key, repository_slug)
         return self.put(url, data=params)
 
+    def get_repo_forkable(self, project_key, repository_slug):
+        """Return whether a Bitbucket Server/Data Center repository is forkable.
+
+        :param project_key: Project key, or a ``~user`` personal repository owner.
+        :param repository_slug: URL-compatible repository identifier.
+        :return: The repository's ``forkable`` flag, or ``None`` when omitted
+            by an older Bitbucket version.
+        """
+        return (self.get_repo(project_key, repository_slug) or {}).get("forkable")
+
+    def set_repo_forkable(self, project_key, repository_slug, forkable):
+        """Enable or disable forking for a Server/Data Center repository.
+
+        The caller requires repository administration permission. This uses the
+        repository ``PUT`` endpoint and preserves the behavior of
+        :meth:`update_repo` for all other repository fields.
+
+        :param project_key: Project key, or a ``~user`` personal repository owner.
+        :param repository_slug: URL-compatible repository identifier.
+        :param forkable: ``True`` to allow forks or ``False`` to prohibit them.
+        :return: Updated repository representation.
+        """
+        if not isinstance(forkable, bool):
+            raise TypeError("forkable must be a boolean")
+        return self.update_repo(project_key, repository_slug, forkable=forkable)
+
+    def enable_repo_forking(self, project_key, repository_slug):
+        """Enable forking for a Server/Data Center repository."""
+        return self.set_repo_forkable(project_key, repository_slug, True)
+
+    def disable_repo_forking(self, project_key, repository_slug):
+        """Disable forking for a Server/Data Center repository."""
+        return self.set_repo_forkable(project_key, repository_slug, False)
+
     def delete_repo(self, project_key, repository_slug):
         """
         Delete a specific repository from a project. This operates based on slug not name which may

atlassian/bitbucket/cloud/repositories/commits.py

@@ -15,15 +15,19 @@ def __init__(self, url, *args, **kwargs):
     def __get_object(self, data):
         return Commit(data, **self._new_session_args)
 
-    def each(self, top=None, q=None, sort=None):
+    def each(self, top=None, q=None, sort=None, include=None, exclude=None, path=None):
         """
         Return the list of commits in this repository.
 
         :param top: string: Hash of commit to get the history for.
-        :param q: string: Query string to narrow down the response.
-                          See https://developer.atlassian.com/bitbucket/api/2/reference/meta/filtering for details.
+        :param q: string: Legacy query expression. Bitbucket Cloud does not
+                          support arbitrary ``q`` filtering for commits; it is
+                          retained for compatibility and passed to the API.
         :param sort: string: Name of a response property to sort results.
                              See https://developer.atlassian.com/bitbucket/api/2/reference/meta/filtering for details.
+        :param include: string: Commit or ref to include in the history.
+        :param exclude: string: Commit or ref to exclude from the history.
+        :param path: string: File path used to filter commits.
 
         :return: A generator for the Commit objects
 
@@ -35,6 +39,12 @@ def each(self, top=None, q=None, sort=None):
             params["sort"] = sort
         if q is not None:
             params["q"] = q
+        if include is not None:
+            params["include"] = include
+        if exclude is not None:
+            params["exclude"] = exclude
+        if path is not None:
+            params["path"] = path
         trailing = True
         if top is not None:
             trailing = False

atlassian/bitbucket/cloud/repositories/repositoryVariables.py

@@ -60,6 +60,29 @@ def each(self, q=None, sort=None):
 
         return
 
+    def get_by_key(self, key):
+        """Return the repository variable with ``key``, or ``None``.
+
+        Bitbucket requires the variable UUID for the item endpoint. This
+        helper follows the paginated collection and returns the object with
+        its UUID, avoiding callers having to implement that lookup themselves.
+        """
+        for variable in self.each(q=f'key="{key}"'):
+            if variable.key == key:
+                return variable
+        return None
+
+    def update_by_key(self, key, value, secured=None):
+        """Update a repository variable by its key and return the object."""
+        variable = self.get_by_key(key)
+        if variable is None:
+            raise ValueError(f"Repository variable with key {key!r} was not found")
+
+        data = {"value": value}
+        if secured is not None:
+            data["secured"] = secured
+        return variable.update(**data)
+
     def get(self, uuid: str):  # type: ignore[override]
         """
         Returns the pipeline with the uuid in this repository.

atlassian/bitbucket/cloud/workspaces/__init__.py

@@ -16,7 +16,7 @@ def __init__(self, url, *args, **kwargs):
     def __get_object(self, data):
         return Workspace(data, **self._new_session_args)
 
-    def each(self, role=None, q=None, sort=None):
+    def each(self, role=None, q=None, sort=None, administrator=None):
         """
         Get all workspaces matching the criteria.
 
@@ -33,6 +33,9 @@ def each(self, role=None, q=None, sort=None):
         :param sort: string (default is None):
                         Name of a response property to sort results.
                         See https://developer.atlassian.com/bitbucket/api/2/reference/meta/filtering for details.
+        :param administrator: bool (default is None):
+                              Filter workspaces by whether the authenticated user
+                              is a workspace administrator.
 
         :return: A generator for the Workspace objects
 
@@ -48,6 +51,8 @@ def each(self, role=None, q=None, sort=None):
         # would produce an invalid request.
         if sort is not None:
             params["sort"] = sort
+        if administrator is not None:
+            params["administrator"] = administrator
         user_workspaces_url = f"{self.url.rsplit('/', 1)[0]}/user/workspaces"
         for workspace_access in self._get_paged(user_workspaces_url, params=params, absolute=True):
             workspace_data = workspace_access.get("workspace", workspace_access)

atlassian/compass/__init__.py

@@ -0,0 +1,109 @@
+"""Atlassian Compass REST API client."""
+
+from ..rest_client import AtlassianRestAPI
+
+
+class Compass(AtlassianRestAPI):
+    """Client for the Compass gateway REST API.
+
+    Compass Cloud exposes these endpoints below ``/gateway/api`` on an
+    Atlassian site. JSON requests use ``data`` or ``json`` as accepted by the
+    base REST client; multipart methods accept a file path or a ``files``
+    mapping.
+    """
+
+    def __init__(self, *args, **kwargs):
+        kwargs["api_root"] = "gateway/api"
+        super().__init__(*args, **kwargs)
+
+    @staticmethod
+    def _multipart_file(filename, files=None):
+        if files is not None:
+            return files, None
+        if filename is None:
+            raise ValueError("filename or files is required")
+        handle = open(filename, "rb")
+        return {"file": handle}, handle
+
+    def send_metric(self, data=None):
+        """Send a metric value to Compass."""
+        return self.post(self.url_joiner(self.api_root, "compass/v1/metrics"), data=data)
+
+    def send_event(self, data=None):
+        """Send a streamlined event to Compass."""
+        return self.post(self.url_joiner(self.api_root, "compass/v1/events"), data=data)
+
+    def get_entitlement_results(self, data=None):
+        """Retrieve entitlement results for a Compass component."""
+        return self.post(self.url_joiner(self.api_root, "compass/v1/entitlements"), data=data)
+
+    def invoke_webhook(self, webhook_id, data=None):
+        """Invoke an inbound Compass webhook."""
+        path = self.url_joiner(self.api_root, f"compass/v1/webhooks/{webhook_id}")
+        return self.post(path, data=data)
+
+    def upload_package_dependencies_lock_file(
+        self, source_id, base_source_url, component_id, filename=None, files=None
+    ):
+        """Upload a package dependency lock file."""
+        multipart, handle = self._multipart_file(filename, files)
+        try:
+            path = self.url_joiner(self.api_root, "compass/v1/package_dependencies/lock_file")
+            params = {
+                "sourceId": source_id,
+                "baseSourceUrl": base_source_url,
+                "componentId": component_id,
+            }
+            return self.put(path, params=params, files=multipart)
+        finally:
+            if handle:
+                handle.close()
+
+    def delete_package_dependencies(self, component_id, source_id):
+        """Delete package dependencies for a component and source."""
+        path = self.url_joiner(self.api_root, f"compass/v1/package_dependencies/lock_file/{component_id}/{source_id}")
+        return self.delete(path)
+
+    def get_forge_app_attachment(self, component_id, forge_app_id, key, not_json_response=True):
+        """Download a Forge app attachment belonging to a component."""
+        path = self.url_joiner(
+            self.api_root, f"compass/v1/component/{component_id}/app/{forge_app_id}/attachment/{key}"
+        )
+        return self.get(path, not_json_response=not_json_response)
+
+    def upload_forge_app_attachment(self, component_id, forge_app_id, key, filename=None, files=None):
+        """Upload a Forge app attachment."""
+        multipart, handle = self._multipart_file(filename, files)
+        try:
+            path = self.url_joiner(
+                self.api_root, f"compass/v1/component/{component_id}/app/{forge_app_id}/attachment/{key}"
+            )
+            return self.put(path, files=multipart)
+        finally:
+            if handle:
+                handle.close()
+
+    def delete_forge_app_attachment(self, component_id, forge_app_id, key):
+        """Delete a Forge app attachment."""
+        path = self.url_joiner(
+            self.api_root, f"compass/v1/component/{component_id}/app/{forge_app_id}/attachment/{key}"
+        )
+        return self.delete(path)
+
+    def upload_component_api_spec(self, component_id, filename=None, files=None):
+        """Upload an OpenAPI specification for a Compass component."""
+        multipart, handle = self._multipart_file(filename, files)
+        try:
+            path = self.url_joiner(self.api_root, f"compass/v1/component/{component_id}/api_specs")
+            return self.put(path, files=multipart)
+        finally:
+            if handle:
+                handle.close()
+
+    def delete_component_api_spec(self, component_id):
+        """Delete the API specification for a Compass component."""
+        path = self.url_joiner(self.api_root, f"compass/v1/component/{component_id}/api_specs")
+        return self.delete(path)
+
+
+__all__ = ["Compass"]