Initial commit with OpenAPI parsing

Author: dbanty

.circleci/config.yml

@@ -0,0 +1,46 @@
+jobs:
+  build:
+    docker:
+    - image: circleci/python:3.8.1
+    steps:
+    - checkout
+    - restore_cache:
+        keys:
+        - v1-dependencies-{{ checksum "poetry.lock" }}
+    - run:
+        command: 'pip install poetry --user --upgrade
+
+          poetry config virtualenvs.in-project true
+
+          poetry config repositories.triaxtec https://pypi.fury.io/triaxtec/
+
+          poetry config http-basic.triaxtec $GEMFURY_PULL_TOKEN $GEMFURY_PULL_TOKEN
+
+          poetry install
+
+          '
+        name: install dependencies
+    - run:
+        command: 'mkdir -p test-reports/safety test-reports/mypy test-reports/pytest
+
+          poetry run black . --check
+
+          poetry run safety check --json > test-reports/safety/results.json
+
+          poetry run mypy openapi_python_client --junit-xml=test-reports/mypy/results.xml
+
+          poetry run pytest --junitxml=test-reports/pytest/results.xml
+
+          '
+        name: run tests
+    - store_test_results:
+        path: test-reports
+    - run:
+        command: poetry run pip uninstall openapi-python-client -y
+        name: Uninstall Package
+    - save_cache:
+        key: v1-dependencies-{{ checksum "poetry.lock" }}
+        paths:
+        - ./.venv
+    working_directory: ~/repo
+version: 2

.gitignore

@@ -0,0 +1,26 @@
+__pycache__/
+build/
+dist/
+*.egg-info/
+.pytest_cache/
+
+# Sphinx documentation
+docs/_build/
+
+# pyenv
+.python-version
+
+# Environments
+.env
+.venv
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# JetBrains
+.idea/
+
+# Terraform
+.terraform

CHANGELOG.md

@@ -0,0 +1,8 @@
+# Changelog
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## 0.1.0 - YYYY-MM-DD
+- Initial Release

README.md

@@ -0,0 +1,26 @@
+# openapi-python-client
+
+## Purpose
+Generate modern Python clients from OpenAPI
+
+## Contribution Guidelines
+ - The project is written to support Python 3.8 and should conform to the [Triax Python Standards](https://triaxtec.atlassian.net/wiki/spaces/EN/pages/499482627/Python+Guidelines).
+ - Any changes should be covered with a unit test and documented in [CHANGELOG.md]
+
+## Release Process
+1. Start a release with Git Flow
+1. Update the version number in `pyproject.toml` with `poetry version <rule>`
+1. Ensure all requirements are pointing to released versions
+1. Add the release date to the new version in [CHANGELOG.md]
+1. Commit and push any changes
+1. Create a pull request from the release branch to master
+1. Get approval from all stakeholders
+1. Ensure all checks pass (e.g. CircleCI)
+1. Open and merge the pull request
+1. Create a tag on the merge commit with the release number
+
+## Contributors 
+ - Dylan Anthony <[email protected]>
+
+
+[CHANGELOG.md]: CHANGELOG.md

mypy.ini

@@ -0,0 +1,7 @@
+[mypy]
+disallow_any_explicit = True
+disallow_any_generics = True
+disallow_untyped_defs = True
+warn_redundant_casts = True
+warn_unused_ignores = True
+strict_equality = True

openapi_python_client/__init__.py

@@ -0,0 +1,26 @@
+""" Generate modern Python clients from OpenAPI """
+import sys
+from typing import Dict
+
+import orjson
+import requests
+
+from .models import OpenAPI
+
+
+def main():
+    """ Generate the client library """
+    url = sys.argv[1]
+    json = _get_json(url)
+    data_dict = _parse_json(json)
+    openapi = OpenAPI.from_dict(data_dict)
+    print(openapi)
+
+
+def _get_json(url) -> bytes:
+    response = requests.get(url)
+    return response.content
+
+
+def _parse_json(json: bytes) -> Dict:
+    return orjson.loads(json)

openapi_python_client/__main__.py

@@ -0,0 +1,3 @@
+from . import main
+
+main()

openapi_python_client/models/__init__.py

@@ -0,0 +1,4 @@
+""" Classes representing the data in the OpenAPI schema """
+
+from .api_info import APIInfo
+from .openapi import OpenAPI

openapi_python_client/models/api_info.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class APIInfo:
+    """ Info about the API """
+
+

openapi_python_client/models/openapi.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Dict, List, Optional
+
+from .properties import Property, property_from_dict
+
+
+class Method(Enum):
+    """ HTTP Methods """
+    GET = "get"
+    POST = "post"
+    PATCH = "patch"
+
+
+class ParameterLocation(Enum):
+    """ The places Parameters can be put when calling an Endpoint """
+    QUERY = "query"
+    PATH = "path"
+
+
+@dataclass
+class Parameter:
+    """ A parameter in an Endpoint """
+    location: ParameterLocation
+    property: Property
+
+    @staticmethod
+    def from_dict(d: Dict, /) -> Parameter:
+        """ Construct a parameter from it's OpenAPI dict form """
+        return Parameter(
+            location=ParameterLocation(d["in"]),
+            property=property_from_dict(
+                name=d["name"],
+                required=d["required"],
+                data=d["schema"],
+            ),
+        )
+
+
+@dataclass
+class Endpoint:
+    """
+    Describes a single endpoint on the server
+    """
+    path: str
+    method: Method
+    description: Optional[str]
+    name: str
+    parameters: List[Parameter]
+    tag: Optional[str] = None
+
+    @staticmethod
+    def get_list_from_dict(d: Dict[str, Dict[str, Dict]], /) -> List[Endpoint]:
+        """ Parse the openapi paths data to get a list of endpoints """
+        endpoints = []
+        for path, path_data in d.items():
+            for method, method_data in path_data.items():
+                parameters: List[Parameter] = []
+                for param_dict in method_data.get("parameters", []):
+                    parameters.append(Parameter.from_dict(param_dict))
+                endpoint = Endpoint(
+                    path=path,
+                    method=Method(method),
+                    description=method_data.get("description"),
+                    name=method_data["operationId"],
+                    parameters=parameters,
+                    tag=method_data.get("tags", [None])[0],
+                )
+                endpoints.append(endpoint)
+        return endpoints
+
+
+@dataclass
+class Schema:
+    """
+    Describes a schema, AKA data model used in requests.
+
+    These will all be converted to dataclasses in the client
+    """
+
+    title: str
+    properties: List[Property]
+    description: str
+
+    @staticmethod
+    def from_dict(d: Dict, /) -> Schema:
+        """ A single Schema from its dict representation """
+        required = set(d.get("required", []))
+        properties: List[Property] = []
+        for key, value in d["properties"].items():
+            properties.append(property_from_dict(name=key, required=key in required, data=value,))
+        return Schema(title=d["title"], properties=properties, description=d.get("description", ""),)
+
+    @staticmethod
+    def dict(d: Dict, /) -> Dict[str, Schema]:
+        """ Get a list of Schemas from an OpenAPI dict """
+        result = {}
+        for data in d.values():
+            s = Schema.from_dict(data)
+            result[s.title] = s
+        return result
+
+
+@dataclass
+class OpenAPI:
+    """ Top level OpenAPI spec """
+
+    title: str
+    description: str
+    version: str
+    security_schemes: Dict
+    schemas: Dict[str, Schema]
+    endpoints: List[Endpoint]
+
+    @staticmethod
+    def from_dict(d: Dict, /) -> OpenAPI:
+        """ Create an OpenAPI from dict """
+        return OpenAPI(
+            title=d["info"]["title"],
+            description=d["info"]["description"],
+            version=d["info"]["version"],
+            endpoints=Endpoint.get_list_from_dict(d["paths"]),
+            schemas=Schema.dict(d["components"]["schemas"]),
+            security_schemes=d["components"]["securitySchemes"],
+        )

openapi_python_client/models/properties.py

@@ -0,0 +1,105 @@
+from dataclasses import dataclass
+from typing import Optional, List, Dict, Union
+
+
+@dataclass
+class Property:
+    """ Describes a single property for a schema """
+
+    name: str
+    required: bool
+
+
+@dataclass
+class StringProperty(Property):
+    """ A property of type str """
+
+    max_length: Optional[int] = None
+    default: Optional[str] = None
+    pattern: Optional[str] = None
+
+
+@dataclass
+class DateTimeProperty(Property):
+    """ A property of type datetime.datetime """
+
+    pass
+
+
+@dataclass
+class FloatProperty(Property):
+    """ A property of type float """
+
+    default: Optional[float] = None
+
+
+@dataclass
+class IntProperty(Property):
+    """ A property of type int """
+
+    default: Optional[int] = None
+
+
+@dataclass
+class BooleanProperty(Property):
+    """ Property for bool """
+
+    pass
+
+
+@dataclass
+class ListProperty(Property):
+    """ Property for list """
+
+    type: Optional[str] = None
+    ref: Optional[str] = None
+
+
+@dataclass
+class EnumProperty(Property):
+    """ A property that should use an enum """
+
+    values: List[str]
+
+
+@dataclass
+class RefProperty(Property):
+    """ A property which refers to another Schema """
+
+    ref: str
+
+
+@dataclass
+class DictProperty(Property):
+    """ Property that is a general Dict """
+
+
+def property_from_dict(
+    name: str, required: bool, data: Dict[str, Union[float, int, str, List[str], Dict[str, str]]]
+) -> Property:
+    if "enum" in data:
+        return EnumProperty(name=name, required=required, values=data["enum"],)
+    if "$ref" in data:
+        ref = data["$ref"].split("/")[-1]
+        return RefProperty(name=name, required=required, ref=ref,)
+    if data["type"] == "string":
+        if "format" not in data:
+            return StringProperty(
+                name=name, default=data.get("default"), required=required, pattern=data.get("pattern"),
+            )
+        elif data["format"] == "date-time":
+            return DateTimeProperty(name=name, required=required,)
+    elif data["type"] == "number":
+        return FloatProperty(name=name, default=data.get("default"), required=required,)
+    elif data["type"] == "integer":
+        return IntProperty(name=name, default=data.get("default"), required=required,)
+    elif data["type"] == "boolean":
+        return BooleanProperty(name=name, required=required,)
+    elif data["type"] == "array":
+        ref = None
+        if "$ref" in data["items"]:
+            ref = data["items"]["$ref"].split("/")[-1]
+        return ListProperty(name=name, required=required, type=data["items"].get("type"), ref=ref,)
+    elif data["type"] == "object":
+        return DictProperty(name=name, required=required,)
+    raise ValueError(f"Did not recognize type of {data}")

openapi_python_client/py.typed

@@ -0,0 +1 @@
+# Marker file for PEP 561