feat: add Artifactory Query Language integration #72
Conversation
There was a problem hiding this comment.
Hello @anancarv and @saper, I am very sorry I took more than a year to review this Ananias 😓
Thank you @saper for your work on the rebase! None of the remarks here are directed towards you, but if you want to work on them more, feel free ^^
I will "request changes" here just to make sure we discuss about it, but I'm overall ok with the PR
| class DomainQueryEnum(str, Enum): | ||
| "Artifactory domain objects to be queried" | ||
| items = "items" | ||
| builds = "builds" | ||
| entries = "entries" |
There was a problem hiding this comment.
If I understand correctly this part of the docs
AQL is constructed as a set of interconnected domains as displayed in the diagram below. You may run queries only on one domain at a time, and this is referred to as the Primary domain of the query.
Currently, the following are supported as primary domains: Item, Build, Entry, Promotion and Release. i.e., your queries may be of the form:items.find(...),builds.find(...),archive.entries.find(...),build.promotions.find(...)orreleases.find(...).
Artifactory actually supports more than just the "items, builds or entries" that they announce. And even more, you cannot just use entries.find() (or at least on our instance it returns "Failedtoparsequery").
FYI I tested all the primary domains they advertise in the quoted docs on our instance, and none of them fail (we have no results for releases and build.promotions but that may be because we never used the feature)
| class Aql(BaseModel): | ||
| "Artifactory Query Language" | ||
| domain: DomainQueryEnum = DomainQueryEnum.items | ||
| find: Optional[Dict[str, Union[str, List[Dict[str, Any]], Dict[str, str]]]] |
There was a problem hiding this comment.
I wonder if you could make a self-referential type here...
The "find" criterion can be a Dict of:
- strings
- dict of strings
- list of find criteria
But after thinking about it I'm not sure you could extract such a type with forward references, because it's not a Pydantic model but a union of many types...
Maybe something like this could work:
AqlFindCriterion = Dict[str, Union[str, Dict[str, str], List["AqlFindCriterion"]]]| @@ -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): | |||
There was a problem hiding this comment.
Maybe this function could become a method in the Aql class, to be used something like this:
aql = Aql(<params>)
aql.to_query() # items.find().include()...| "Artifactory Query Language support" | ||
| _uri = "search/aql" | ||
|
|
||
| def query(self, aql_object: Aql) -> List[Dict[str, Union[str, List]]]: |
There was a problem hiding this comment.
But then I wondered about the API we expose... Should the user create an Aql object, and pass it to the method, or should the method take parameters like find, include... that would individually be typed?
I don't really know the purpose of the Aql object, but maybe you thought about it and have a reason that I don't know ^^
| response_content: List[Dict[str, Union[str, List]]] = response.json()[ | ||
| "results" | ||
| ] |
There was a problem hiding this comment.
You should parse the response here, with parse_from_obj(List[Dict[str, Union[str, List]]], response.json()["results"] (and you won't have to type it because parse_from_obj is cleverly typed 😉)
There was a problem hiding this comment.
Wait, actually should we return only the results key? The user might need the other keys, for example to know how many results were returned.
|
|
||
|
|
||
| @responses.activate | ||
| def test_aql_fail_baq_query(): |
There was a problem hiding this comment.
Wooops, that should read ^^
| def test_aql_fail_baq_query(): | |
| def test_aql_fail_bad_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 += "()" |
There was a problem hiding this comment.
To keep it consistent with the other parts of the function (and simpler to understand), maybe move the .find to the if block:
| 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 += "()" | |
| aql_query_text = f"{aql_object.domain}" | |
| if aql_object.find: | |
| aql_query_text += f".find({json.dumps(aql_object.find)})" | |
| else: | |
| aql_query_text += ".find()" |
|
|
||
| if aql_object.include: | ||
| format_include = ( | ||
| json.dumps(aql_object.include).replace("[", "").replace("]", "") |
There was a problem hiding this comment.
Not sure about this, what would happen if there is a [ in one of the included fields? It would probably not happen, I know, but maybe just removing them as the first and last characters would be safer?
| if aql_object.offset: | ||
| aql_query_text += f".offset({aql_object.offset})" | ||
|
|
||
| if aql_object.limit: | ||
| aql_query_text += f".limit({aql_object.limit})" |
There was a problem hiding this comment.
Be careful with this sort of comparison, if something is not the same as if something is not None! In the first case you will not enter the if if the user sends a 0!
| @@ -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): | |||
There was a problem hiding this comment.
Also I think this method could benefit from being more unit tested, to make sure we generate the correct query whatever the combination of parameters we give. I can write those if you want 😉
This is a rebase of #30