Skip to content

Commit

Permalink
Improving Code Comments
Browse files Browse the repository at this point in the history
  • Loading branch information
luolingchun committed Jun 2, 2023
1 parent b2910e4 commit e3b928a
Show file tree
Hide file tree
Showing 7 changed files with 419 additions and 218 deletions.
65 changes: 40 additions & 25 deletions flask_openapi3/blueprint.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,49 +35,63 @@ def __init__(
Arguments:
name: The name of the blueprint. It Will be prepared to each endpoint name.
import_name: The name of the blueprint package, usually
``__name__``. This helps locate the ``root_path`` for the blueprint.
abp_tags: APIBlueprint tags for every api
abp_security: APIBlueprint security for every api
abp_responses: API responses, should be BaseModel, dict or None.
doc_ui: Add openapi document UI(swagger, rapidoc and redoc). Defaults to True.
import_name: The name of the blueprint package, usually ``__name__``.
This helps locate the ``root_path`` for the blueprint.
abp_tags: APIBlueprint tags for every API.
abp_security: APIBlueprint security for every API.
abp_responses: API responses should be either a subclass of BaseModel, a dictionary, or None.
doc_ui: Enable OpenAPI document UI (Swagger UI, Redoc and Rapidoc). Defaults to True.
operation_id_callback: Callback function for custom operation_id generation.
Receives name (str), path (str) and method (str) parameters.
Defaults to `get_operation_id_for_path` from utils
Receives name (str), path (str) and method (str) parameters.
Defaults to `get_operation_id_for_path` from utils
**kwargs: Flask Blueprint kwargs
"""
super(APIBlueprint, self).__init__(name, import_name, **kwargs)

# Initialize instance variables
self.paths: Dict = dict()
self.components_schemas: Dict = dict()
self.tags: List[Tag] = []
self.tag_names: List[str] = []

# Set values from arguments or default values
self.abp_tags = abp_tags or []
self.abp_security = abp_security or []
self.abp_responses = abp_responses or {}
self.doc_ui = doc_ui

# Set the operation ID callback function
self.operation_id_callback: Callable = operation_id_callback

def register_api(self, api: "APIBlueprint") -> None:
"""Register a nested APIBlueprint"""

# Check if the APIBlueprint is being registered on itself
if api is self:
raise ValueError("Cannot register a api blueprint on itself")

# Merge tags from the nested APIBlueprint
for tag in api.tags:
if tag.name not in self.tag_names:
self.tags.append(tag)

# Merge paths from the nested APIBlueprint
for path_url, path_item in api.paths.items():
trail_slash = path_url.endswith("/")
# merge url_prefix and new api blueprint path url

# Merge url_prefix and new API blueprint path url
uri = self.url_prefix.rstrip("/") + "/" + path_url.lstrip("/") if self.url_prefix else path_url
# strip the right slash

# Strip the right slash
if not trail_slash:
uri = uri.rstrip("/")

self.paths[uri] = path_item

# Merge component schemas from the nested APIBlueprint
self.components_schemas.update(**api.components_schemas)

# Register the nested APIBlueprint as a blueprint
self.register_blueprint(api)

def _do_decorator(
Expand All @@ -102,7 +116,8 @@ def _do_decorator(
method: str = HTTPMethod.GET
) -> Tuple[Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel], Type[BaseModel]]:
"""
Collect openapi specification information
Collects OpenAPI specification information for Flask routes and view functions.
Arguments:
rule: Flask route
func: Flask view_func
Expand All @@ -113,7 +128,7 @@ def _do_decorator(
operation_id: Unique string used to identify the operation.
extra_form: Extra information describing the request body(application/form).
extra_body: Extra information describing the request body(application/json).
responses: API responses, should be BaseModel, dict or None.
responses: API responses should be either a subclass of BaseModel, a dictionary, or None.
extra_responses: Extra information for responses.
deprecated: Declares this operation to be deprecated.
security: A declaration of which security mechanisms can be used for this operation.
Expand All @@ -126,54 +141,54 @@ def _do_decorator(
responses = {}
if extra_responses is None:
extra_responses = {}
# global response combine api responses
# Global response: combine API responses
combine_responses = deepcopy(self.abp_responses)
combine_responses.update(**responses)
# create operation
# Create operation
operation = get_operation(
func,
summary=summary,
description=description,
openapi_extensions=openapi_extensions
)
# set external docs
# Set external docs
operation.externalDocs = external_docs
# Unique string used to identify the operation.
operation.operationId = operation_id or self.operation_id_callback(
name=func.__name__, path=rule, method=method
)
# only set `deprecated` if True otherwise leave it as None
# Only set `deprecated` if True, otherwise leave it as None
operation.deprecated = deprecated
# add security
# Add security
if security is None:
security = []
operation.security = security + self.abp_security or None
# add servers
# Add servers
operation.servers = servers
# store tags
# Store tags
tags = tags + self.abp_tags if tags else self.abp_tags
parse_and_store_tags(tags, self.tags, self.tag_names, operation)
# parse parameters
# Parse parameters
header, cookie, path, query, form, body = parse_parameters(
func,
extra_form=extra_form,
extra_body=extra_body,
components_schemas=self.components_schemas,
operation=operation
)
# parse response
# Parse response
get_responses(combine_responses, extra_responses, self.components_schemas, operation)
# /pet/<petId> --> /pet/{petId}
# Convert route parameter format from /pet/<petId> to /pet/{petId}
uri = re.sub(r"<([^<:]+:)?", "{", rule).replace(">", "}")
trail_slash = uri.endswith("/")
# merge url_prefix and uri
# Merge url_prefix and uri
uri = self.url_prefix.rstrip("/") + "/" + uri.lstrip("/") if self.url_prefix else uri
if not trail_slash:
uri = uri.rstrip("/")
# parse method
# Parse method
parse_method(uri, method, self.paths, operation)
return header, cookie, path, query, form, body
else:
# parse parameters
# Parse parameters
header, cookie, path, query, form, body = parse_parameters(func, doc_ui=False)
return header, cookie, path, query, form, body
24 changes: 15 additions & 9 deletions flask_openapi3/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,28 +10,34 @@
from .markdown import openapi_to_markdown


@click.command(name='openapi')
@click.option('--output', '-o', type=click.Path(), help='The output file path.')
@click.option('--format', '-f', type=click.Choice(['json', 'yaml', 'markdown']), help='The output file format.')
@click.option('--indent', '-i', type=int, help='The indentation for JSON dumps.')
@click.option('--ensure_ascii', '-a', is_flag=True, help='Ensure ASCII characters or not. Defaults to False.')
@click.command(name="openapi")
@click.option("--output", "-o", type=click.Path(), help="The output file path.")
@click.option("--format", "-f", type=click.Choice(["json", "yaml", "markdown"]), help="The output file format.")
@click.option("--indent", "-i", type=int, help="The indentation for JSON dumps.")
@click.option("--ensure_ascii", "-a", is_flag=True, help="Ensure ASCII characters or not. Defaults to False.")
@with_appcontext
def openapi_command(output, format, indent, ensure_ascii):
"""Export the OpenAPI Specification to console or a file"""
if hasattr(current_app, 'api_doc'):

# Check if the current app has an api_doc attribute
if hasattr(current_app, "api_doc"):
obj = current_app.api_doc
if format == 'yaml':

# Generate the OpenAPI Specification based on the specified format
if format == "yaml":
try:
import yaml # type: ignore
except ImportError:
raise ImportError("pyyaml must be installed.")
openapi = yaml.safe_dump(obj, allow_unicode=True)
elif format == 'markdown':
elif format == "markdown":
openapi = openapi_to_markdown(obj)
else:
openapi = json.dumps(obj, indent=indent, ensure_ascii=ensure_ascii)

# Save the OpenAPI Specification to a file if the output path is provided
if output:
with open(output, 'w', encoding='utf8') as f:
with open(output, "w", encoding="utf8") as f:
f.write(openapi)
click.echo(f"Saved to {output}.")
else:
Expand Down
Loading

0 comments on commit e3b928a

Please sign in to comment.