Merge pull request #29 from johmau/add_components_schemas_parsing

feat: add #/components/schemas parsing to retrieve explicit schema

Author: manchenkoff

src/openapi_parser/builders/__init__.py

@@ -12,6 +12,7 @@
 from .path import PathBuilder
 from .oauth_flow import OAuthFlowBuilder
 from .security import SecurityBuilder
+from .schemas import SchemasBuilder
 
 __all__ = [
     "InfoBuilder",
@@ -28,4 +29,5 @@
     "PathBuilder",
     "OAuthFlowBuilder",
     "SecurityBuilder",
+    "SchemasBuilder",
 ]

src/openapi_parser/builders/schemas.py

@@ -0,0 +1,21 @@
+import logging
+
+from ..specification import Schema
+from . import SchemaFactory
+
+
+logger = logging.getLogger(__name__)
+
+class SchemasBuilder:
+    schema_factory: SchemaFactory
+
+    def __init__(self, schema_factory: SchemaFactory) -> None:
+        self.schema_factory = schema_factory
+
+    def build_collection(self, schemas: dict) -> dict[str, Schema]:
+        logger.debug(f"Schemas parsing: {schemas.keys()}")
+
+        return {
+            key : self.schema_factory.create(value)
+            for key, value in schemas.items()
+        }

src/openapi_parser/parser.py

@@ -16,20 +16,23 @@ class Parser:
     external_doc_builder: ExternalDocBuilder
     path_builder: PathBuilder
     security_builder: SecurityBuilder
+    schemas_builder: SchemasBuilder
 
     def __init__(self,
                  info_builder: InfoBuilder,
                  server_builder: ServerBuilder,
                  tags_builder: TagBuilder,
                  external_doc_builder: ExternalDocBuilder,
                  path_builder: PathBuilder,
-                 security_builder: SecurityBuilder) -> None:
+                 security_builder: SecurityBuilder,
+                 schemas_builder: SchemasBuilder) -> None:
         self.info_builder = info_builder
         self.server_builder = server_builder
         self.tag_builder = tags_builder
         self.external_doc_builder = external_doc_builder
         self.path_builder = path_builder
         self.security_builder = security_builder
+        self.schemas_builder = schemas_builder
 
     def load_specification(self, data: dict) -> Specification:
         """Load OpenAPI Specification object from a file or a remote URI.
@@ -67,6 +70,9 @@ def load_specification(self, data: dict) -> Specification:
         if data.get('components') and data['components'].get('securitySchemes'):
             attrs["security_schemas"] = self.security_builder.build_collection(data['components']['securitySchemes'])
 
+        if data.get('components') and data['components'].get('schemas'):
+            attrs["schemas"] = self.schemas_builder.build_collection(data['components']['schemas'])
+
         logger.debug("Specification parsed successfully")
 
         return Specification(**attrs)
@@ -83,6 +89,7 @@ def _create_parser() -> Parser:
     content_builder = ContentBuilder(schema_factory)
     header_builder = HeaderBuilder(schema_factory)
     parameter_builder = ParameterBuilder(schema_factory)
+    schemas_builder = SchemasBuilder(schema_factory)
     response_builder = ResponseBuilder(content_builder, header_builder)
     request_builder = RequestBuilder(content_builder)
     operation_builder = OperationBuilder(response_builder,
@@ -98,7 +105,8 @@ def _create_parser() -> Parser:
                   tag_builder,
                   external_doc_builder,
                   path_builder,
-                  security_builder)
+                  security_builder,
+                  schemas_builder)
 
 
 def parse(uri: str) -> Specification:

src/openapi_parser/specification.py

@@ -250,5 +250,6 @@ class Specification:
     tags: list[Tag] = field(default_factory=list)
     security_schemas: dict[str, Security] = field(default_factory=dict)
     security: list[dict[str, Any]] = field(default_factory=list)
+    schemas: dict[str, Schema] = field(default_factory=dict)
     external_docs: Optional[ExternalDoc] = None
     paths: list[Path] = field(default_factory=list)

tests/openapi_fixture.py

@@ -157,6 +157,100 @@ def create_specification() -> Specification:
         ),
     }
 
+    schemas = {
+        "BadRequestError" : Object(
+            type=DataType.OBJECT,
+            required=["code", "error"],
+            properties=[
+                Property(
+                    name="code",
+                    schema=Integer(
+                        type=DataType.INTEGER,
+                        example=1044,
+                        description="Internal error code",
+                    )
+                ),
+                Property(
+                    name="error",
+                    schema=String(
+                        type=DataType.STRING,
+                        example="Invalid user id value",
+                        description="Error details",
+                    )
+                ),
+            ],
+        ),
+        "InternalServerError": Object(
+            type=DataType.OBJECT,
+            required=["code", "error"],
+            properties=[
+                Property(
+                    name="code",
+                    schema=Integer(
+                        type=DataType.INTEGER,
+                        example=1,
+                        description="Internal error code",
+                    )
+                ),
+                Property(
+                    name="error",
+                    schema=String(
+                        type=DataType.STRING,
+                        example="Unexpected server error",
+                        description="Error details",
+                    )
+                ),
+            ],
+        ),
+        "UUIDObject": Object(
+            type=DataType.OBJECT,
+            required=["uuid"],
+            properties=[
+                Property(
+                    name="uuid",
+                    schema=String(
+                        type=DataType.STRING,
+                        format=StringFormat.UUID,
+                        example="12345678-1234-5678-1234-567812345678",
+                        description="Unique object id",
+                    )
+                ),
+            ],
+        ),
+        "User": Object(
+            type=DataType.OBJECT,
+            required=["uuid","login", "email"],
+            properties=[
+                Property(
+                    name="uuid",
+                    schema=String(
+                        type=DataType.STRING,
+                        format=StringFormat.UUID,
+                        example="12345678-1234-5678-1234-567812345678",
+                        description="Unique object id",
+                    )
+                ),
+                Property(
+                    name="login",
+                    schema=String(
+                        type=DataType.STRING,
+                        example="super-admin",
+                        description="User login or nickname",
+                    )
+                ),
+                Property(
+                    name="email",
+                    schema=String(
+                        type=DataType.STRING,
+                        format=StringFormat.EMAIL,
+                        example="user@mail.com",
+                        description="User E-mail address",
+                    )
+                ),
+            ],
+        )
+    }
+
     security = [
         {"Basic": []}
     ]
@@ -300,4 +394,5 @@ def create_specification() -> Specification:
                          tags=tag_list,
                          paths=path_list,
                          security_schemas=security_schemes,
-                         security=security)
+                         security=security,
+                         schemas=schemas)

tests/test_parser.py

@@ -43,13 +43,15 @@ def test_load_specification(swagger_specification: Specification) -> None:
     external_doc_builder = _create_builder_mock(swagger_specification.external_docs)
     path_builder = _create_list_builder_mock(swagger_specification.paths)
     security_builder = _create_collection_builder_mock(swagger_specification.security_schemas)
+    schemas_builder = _create_collection_builder_mock(swagger_specification.schemas)
 
     parser = Parser(info_builder,
                     server_list_builder,
                     tag_list_builder,
                     external_doc_builder,
                     path_builder,
-                    security_builder)
+                    security_builder,
+                    schemas_builder)
 
     swagger_json = OpenAPIResolver(SWAGGER_JSON_FILEPATH).resolve()