Skip to content

Commit

Permalink
operations: List...
Browse files Browse the repository at this point in the history 
Adjusted the Reduce class to handle a list of operations rather than a dict of path and methods.

A string or pattern can be used in the list to match against operationIds, or a tuple with a string or pattern as the first element representing the path and a list of strings representing HTTP verbs as the second element will match against the path/methods.

Also, in any case that a path is not culled, the non-operation keys are kept intact to maintain defaults and overrides.

The MSGraph test was updated to accommodate the annotation change as well as to handle a few validation fixes.
  • Loading branch information
@ggpwnkthx
ggpwnkthx committed Sep 28, 2023
1 parent e574996 commit 21b8dfb
Show file tree
Hide file tree
Showing 2 changed files with 77 additions and 87 deletions.
88 changes: 41 additions & 47 deletions aiopenapi3/extra.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
from re import Pattern
from typing import Dict, List, Union
from typing import List, Union, Optional, Tuple
import logging
import re

Expand All @@ -13,55 +12,50 @@ class Reduce(Document, Init):

log = logging.getLogger("aiopenapi3.extra.Reduce")

def __init__(self, operations: Dict[Union[str, Pattern], List[Union[str, Pattern]]]) -> None:
"""
:param operations: paths/methods to reduce to
"""
self.operations: List[Union[str, Pattern]] = operations
def __init__(
self,
operations: List[
Union[Tuple[Union[re.Pattern, str], Optional[List[Union[re.Pattern, str]]]], Union[re.Pattern, str]]
],
) -> None:
self.operations = operations
super().__init__()

def _reduced_paths(self, ctx: "Document.Context") -> dict:
reduced_paths = {}
for path_key, path_value in ctx.document["paths"].items():
# Extracting Non-Operation Objects
non_op_objects = {
key: val
for key, val in path_value.items()
if key in {"summary", "description", "servers", "parameters"}
}

for operation_key, operation_value in path_value.items():
if operation_key in non_op_objects: # Skip if the key is a Non-Operation Object
continue

if not isinstance(operation_value, dict):
continue

for pattern, operation_patterns in self.operations.items():
# If pattern is None, look for operationId in operation_patterns
if pattern is None and operation_patterns is not None:
operation_id = operation_value.get("operationId", "")
if any(
op_pattern == operation_id
or (isinstance(op_pattern, Pattern) and re.match(op_pattern, operation_id))
for op_pattern in operation_patterns
):
reduced_paths.setdefault(path_key, {}).update(non_op_objects)
reduced_paths[path_key][operation_key] = operation_value

else:
if (
(isinstance(pattern, str) and pattern == path_key)
or (isinstance(pattern, Pattern) and re.match(pattern, path_key))
) and any(
op_pattern == operation_key
or (isinstance(op_pattern, Pattern) and re.match(op_pattern, operation_key))
for op_pattern in operation_patterns or [operation_key]
reduced = {}
keep_keys = {"summary", "description", "servers", "parameters"}
for operation in self.operations:
if isinstance(operation, (str, re.Pattern)):
for path_key, path_value in ctx.document["paths"].items():
for operation_key, operation_value in path_value.items():
if "operationId" in operation_value and (
(isinstance(operation, str) and operation == operation_value["operationId"])
or (
isinstance(operation, re.Pattern)
and re.match(operation, operation_value["operationId"])
)
):
reduced_paths.setdefault(path_key, {}).update(non_op_objects)
reduced_paths[path_key][operation_key] = operation_value

return reduced_paths
if path_key not in reduced:
reduced[path_key] = {k: v for k, v in path_value.items() if k in keep_keys}
reduced[path_key][operation_key] = operation_value
else:
pattern, operation_patterns = operation
for path_key in ctx.document["paths"].keys():
if (isinstance(pattern, str) and pattern == path_key) or (
isinstance(pattern, re.Pattern) and re.match(pattern, path_key)
):
reduced[path_key] = {
k: v
for k, v in ctx.document["paths"][path_key].items()
if k in keep_keys
or not operation_patterns
or any(
(isinstance(op_pattern, str) and op_pattern == k)
or (isinstance(op_pattern, re.Pattern) and re.match(op_pattern, k))
for op_pattern in operation_patterns
)
}
return reduced

def parsed(self, ctx: "Document.Context") -> "Document.Context":
"""Parse the given context."""
Expand Down
76 changes: 36 additions & 40 deletions tests/extra_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@

from aiopenapi3 import OpenAPI
from aiopenapi3.loader import FileSystemLoader
from aiopenapi3.extra import Cull, Reduce
from aiopenapi3.extra import Cull, Init, Reduce, Document
from typing import Dict


class PetStoreReduced(Reduce):
Expand All @@ -22,20 +23,37 @@ def __init__(self):
class MSGraph:
def __init__(self):
super().__init__(
operations={
"/me/profile": None,
re.compile(r"/me/sendMail.*"): None,
}
operations=[
("/me/profile", None),
(re.compile(r"/me/sendMail.*"), None),
"accessReviewDecisions.accessReviewDecision.ListAccessReviewDecision",
re.compile(r"drives.drive.items.driveItem.permissions.permission*"),
]
)

def parsed(self, ctx):
@staticmethod
def _remove_parameter(document, path, parameter_name):
if document["paths"].get(path, {}).get("parameters"):
document["paths"][path]["parameters"] = [
p for p in document["paths"][path]["parameters"] if p.get("name", "") != parameter_name
]

@staticmethod
def _drop_required(schema: Dict, requirement: str) -> None:
if "required" in schema:
schema["required"] = [i for i in schema["required"] if i != requirement]
if not schema["required"]:
del schema["required"]

def parsed(self, ctx: "Document.Context") -> "Document.Context":
# Drop massive unnecessary discriminator
del ctx.document["components"]["schemas"]["microsoft.graph.entity"]["discriminator"]

# Run standard reduction process
ctx = super().parsed(ctx)

# Fix invalids
# Remove superfluous parameters
self._remove_parameter(ctx.document, "/applications(appId='{appId}')", "uniqueName")
self._remove_parameter(ctx.document, "/applications(uniqueName='{uniqueName}')", "appId")
# Fix parameter names
for operation in ctx.document.get("paths", {}).values():
for details in operation.values():
# Check if parameters exist for this operation
Expand All @@ -46,43 +64,21 @@ def parsed(self, ctx):
# Check if description matches the desired format
if description.strip() == "Usage: on='{on}'":
parameter["name"] = "on"

# Drop requirement for @odata.type since it's not actually required
for schema in ctx.document["components"]["schemas"].values():
if "required" in schema:
schema["required"] = [i for i in schema["required"] if i != "@odata.type"]
if not schema["required"]:
del schema["required"]
if "content" in parameter.keys():
parameter["schema"] = parameter["content"].get("application/json", {}).get("schema", {})
del parameter["content"]
# Drop requirement for @odata.type since it's not actually enforced
for schema in ctx.document.get("components", {}).get("schemas", {}).values():
if isinstance(schema, dict):
self._drop_required(schema, "@odata.type")
for s in schema.get("allOf", []):
if "required" in s:
s["required"] = [i for i in s["required"] if i != "@odata.type"]
if not s["required"]:
del s["required"]

ctx.document.setdefault("security", []).append({"token": []})
ctx.document.setdefault("components", {}).setdefault("securitySchemes", {}).setdefault(
"token",
{"type": "http", "scheme": "bearer", "bearerFormat": "JWT"},
)

# Rebuild Tags
ctx.document["tags"] = [
{"name": tag}
for tag in set(
tag
for operations in ctx.document.get("paths", {}).values()
for details in operations.values()
if isinstance(details, dict)
for tag in details.get("tags", [])
)
]

self._drop_required(s, "@odata.type")
return ctx


class MSGraphCulled(MSGraph, Cull):
pass
def paths(self, ctx: "Init.Context") -> "Init.Context":
return ctx


class MSGraphReduced(MSGraph, Reduce):
Expand Down

0 comments on commit 21b8dfb

Please sign in to comment.