feat: add Artifactory Query Language integration

README.md

@@ -33,6 +33,7 @@ This library enables you to manage Artifactory resources such as users, groups,
     + [Copy artifact to a new location](#copy-artifact-to-a-new-location)
     + [Move artifact to a new location](#move-artifact-to-a-new-location)
     + [Delete an artifact](#delete-an-artifact)
+  * [Artifactory Query Language](#artifactory-query-language)
   * [Contributing](#contributing)
 
 <!-- tocstop -->
@@ -320,6 +321,26 @@ artifact = art.artifacts.move("<CURRENT_ARTIFACT_PATH_IN_ARTIFACTORY>","<NEW_ART
 art.artifacts.delete("<ARTIFACT_PATH_IN_ARTIFACTORY>")
 ```
 
+### Artifactory Query Language
+You can use [Artifactory Query Language](https://www.jfrog.com/confluence/display/RTF/Artifactory+Query+Language) to uncover data related to the artifacts and builds stored within Artifactory
+```python
+from pyartifactory import Artifactory
+from pyartifactory.models import Aql
+
+art = Artifactory(url="ARTIFACTORY_URL", auth=('USERNAME','PASSWORD_OR_API_KEY'))
+
+# Create an Aql object with your query parameters
+aql_obj = Aql(**{
+  "domain":"items", 
+  "find":{"name" : {"$match":"*.jar"}}, 
+  "sort": { "$asc" : ["repo","name"] }, 
+  "limit": 100
+})
+
+result = art.aql.query(aql_obj)
+>> print(result)
+[{'repo': 'my-repo', 'path': 'my/path', 'name': 'test.jar', 'type': 'file', 'size': 1111, 'created': 'some-date', 'created_by': 'some-date', 'modified': 'some-data', 'modified_by': 'some-user', 'updated': 'some-data'}]
+```
 
 ### Contributing
 Please read the [Development - Contributing](./CONTRIBUTING.md) guidelines.

pyartifactory/__init__.py

@@ -8,6 +8,7 @@
     ArtifactoryRepository,
     ArtifactoryArtifact,
     ArtifactoryPermission,
+    ArtifactoryAql,
     Artifactory,
     AccessTokenModel,
 )

pyartifactory/exception.py

@@ -49,3 +49,7 @@ class PropertyNotFoundException(ArtifactoryException):
 
 class InvalidTokenDataException(ArtifactoryException):
     """The token contains invalid data."""
+
+
+class AqlException(ArtifactoryException):
+    """AQL search failed"""

pyartifactory/models/__init__.py

@@ -16,6 +16,7 @@
     SimpleRepository,
 )
 
+from .aql import Aql
 from .artifact import (
     ArtifactPropertiesResponse,
     ArtifactStatsResponse,

pyartifactory/models/aql.py

@@ -0,0 +1,28 @@
+"Artifactory queries"
+from typing import List, Dict, Optional, Union, Any
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class SortTypesEnum(str, Enum):
+    "Order of query results"
+    asc = "$asc"
+    desc = "$desc"
+
+
+class DomainQueryEnum(str, Enum):
+    "Artifactory domain objects to be queried"
+    items = "items"
+    builds = "builds"
+    entries = "entries"
+
+
+class Aql(BaseModel):
+    "Artifactory Query Language"
+    domain: DomainQueryEnum = DomainQueryEnum.items
+    find: Optional[Dict[str, Union[str, List[Dict[str, Any]], Dict[str, str]]]]
+    include: Optional[List[str]]
+    sort: Optional[Dict[SortTypesEnum, List[str]]] = None
+    offset: Optional[int] = None
+    limit: Optional[int] = None

pyartifactory/objects.py

@@ -1,6 +1,7 @@
 """
 Definition of all artifactory objects.
 """
+import json
 import warnings
 import logging
 import os
@@ -24,6 +25,7 @@
     PermissionNotFoundException,
     InvalidTokenDataException,
     PropertyNotFoundException,
+    AqlException,
     ArtifactNotFoundException,
 )
 
@@ -48,6 +50,7 @@
     User,
     Permission,
     SimplePermission,
+    Aql,
     ArtifactPropertiesResponse,
     ArtifactStatsResponse,
     ArtifactInfoResponse,
@@ -73,6 +76,7 @@ def __init__(
         self.repositories = ArtifactoryRepository(self.artifactory)
         self.artifacts = ArtifactoryArtifact(self.artifactory)
         self.permissions = ArtifactoryPermission(self.artifactory)
+        self.aql = ArtifactoryAql(self.artifactory)
 
 
 class ArtifactoryObject:
@@ -950,3 +954,53 @@ def delete(self, artifact_path: str) -> None:
         artifact_path = artifact_path.lstrip("/")
         self._delete(f"{artifact_path}")
         logger.debug("Artifact %s successfully deleted", artifact_path)
+
+
+def create_aql_query(aql_object: Aql):
+    "Create Artifactory query"
+    aql_query_text = f"{aql_object.domain}.find"
+
+    if aql_object.find:
+        aql_query_text += f"({json.dumps(aql_object.find)})"
+    else:
+        aql_query_text += "()"
+
+    if aql_object.include:
+        format_include = (
+            json.dumps(aql_object.include).replace("[", "").replace("]", "")
+        )
+        aql_query_text += f".include({format_include})"
+
+    if aql_object.sort:
+        sort_key = list(aql_object.sort.keys())[0]
+        sort_value = json.dumps(aql_object.sort[sort_key])
+        aql_query_text += f'.sort({{"{sort_key.value}": {sort_value}}})'
+
+    if aql_object.offset:
+        aql_query_text += f".offset({aql_object.offset})"
+
+    if aql_object.limit:
+        aql_query_text += f".limit({aql_object.limit})"
+
+    return aql_query_text
+
+
+class ArtifactoryAql(ArtifactoryObject):
+    "Artifactory Query Language support"
+    _uri = "search/aql"
+
+    def query(self, aql_object: Aql) -> List[Dict[str, Union[str, List]]]:
+        "Send Artifactory query"
+        aql_query = create_aql_query(aql_object)
+        try:
+            response = self._post(f"api/{self._uri}", data=aql_query)
+            response_content: List[Dict[str, Union[str, List]]] = response.json()[
+                "results"
+            ]
+            logging.debug("Successful query")
+            return response_content
+        except requests.exceptions.HTTPError as error:
+            raise AqlException(
+                "Bad Aql Query: please check your parameters."
+                "Doc: https://www.jfrog.com/confluence/display/RTF/Artifactory+Query+Language"
+            ) from error

pyproject.toml

@@ -34,7 +34,7 @@ tests
 
 [tool.pylint.messages_control]
 disable = """
-bad-continuation,line-too-long,too-few-public-methods,import-error,too-many-instance-attributes
+bad-continuation,line-too-long,too-few-public-methods,import-error,too-many-instance-attributes,too-many-lines
 """
 
 [tool.pylint.basic]

tests/test_aql.py

@@ -0,0 +1,55 @@
+import pytest
+import responses
+
+from pyartifactory import ArtifactoryAql
+from pyartifactory.exception import AqlException
+from pyartifactory.models import AuthModel, Aql
+
+URL = "http://localhost:8080/artifactory"
+AUTH = ("user", "password_or_apiKey")
+AQL_RESPONSE = {
+    "results": [
+        {
+            "repo": "libs-release-local",
+            "path": "org/jfrog/artifactory",
+            "name": "artifactory.war",
+            "type": "item type",
+            "size": "75500000",
+            "created": "2015-01-01T10:10;10",
+            "created_by": "Jfrog",
+            "modified": "2015-01-01T10:10;10",
+            "modified_by": "Jfrog",
+            "updated": "2015-01-01T10:10;10",
+        }
+    ],
+    "range": {"start_pos": 0, "end_pos": 1, "total": 1},
+}
+
+
+@responses.activate
+def test_aql_success():
+    responses.add(
+        responses.POST, f"{URL}/api/search/aql", json=AQL_RESPONSE, status=200
+    )
+
+    artifactory_aql = ArtifactoryAql(AuthModel(url=URL, auth=AUTH))
+    aql_obj = Aql(**{"find": {"repo": {"$eq": "libs-release-local"}}})
+    result = artifactory_aql.query(aql_obj)
+    assert result == AQL_RESPONSE["results"]
+
+
+@responses.activate
+def test_aql_fail_baq_query():
+    responses.add(
+        responses.POST, f"{URL}/api/search/aql", json=AQL_RESPONSE, status=400
+    )
+
+    artifactory_aql = ArtifactoryAql(AuthModel(url=URL, auth=AUTH))
+    aql_obj = Aql(
+        include=["artifact", "artifact.module", "artifact.module.build"],
+        sort={"$asc": ["remote_downloaded"]},
+        limit=100,
+    )
+
+    with pytest.raises(AqlException):
+        artifactory_aql.query(aql_obj)