0.2.0

CHANGELOG.md

@@ -10,6 +10,8 @@
 ### 0.2.0
 
 - Improved CLI output
+- Code organisation is now sensible and not just one giant file
 - Now supports an openapi spec generated from a dotnet project (`Microsoft.OpenApi.Models`)
-- async client support
-- HTTP Bearer authentication support
+- async client support  fully working
+- HTTP Bearer support
+- HTTP Basic support

README.md

@@ -47,7 +47,7 @@ clientele generate -f path/to/file.json -o output/
 ### Async Client
 
 ```sh
-clientele generate -f path/to/file.json -o output/ --async t
+clientele generate -f path/to/file.json -o output/ --asyncio t
 ```
 
 ## Authentication
@@ -60,5 +60,8 @@ Then clientele will provide you information on the environment variables you nee
 make this work during the generation. For example:
 
 ```sh
-[info ] Generated HTTP Bearer auth, use with this environment variable to use: {EXAMPLE_BEARER_AUTH}
+Please set
+* MY_CLIENT_AUTH_USER_KEY
+* MY_CLIENT_AUTH_PASS_KEY
+environment variable to use basic authentication
 ```

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "clientele"
-version = "0.1.0"
+version = "0.2.0"
 description = "Typed API Clients from OpenAPI specs"
 authors = ["Paul Hallett <paulandrewhallett@gmail.com>"]
 license = "MIT"

src/generator.py

@@ -3,6 +3,7 @@
 from openapi_core import Spec
 
 from src.generators.clients import ClientsGenerator
+from src.generators.http import HTTPGenerator
 from src.generators.schemas import SchemasGenerator
 from src.settings import TEMPLATE_ROOT
 
@@ -16,12 +17,19 @@ class Generator:
     asyncio: bool
     schemas_generator: SchemasGenerator
     clients_generator: ClientsGenerator
+    http_generator: HTTPGenerator
     output_dir: str
 
     def __init__(self, spec: Spec, output_dir: str, asyncio: bool) -> None:
         self.schemas_generator = SchemasGenerator(spec=spec, output_dir=output_dir)
         self.clients_generator = ClientsGenerator(
-            spec=spec, output_dir=output_dir, schemas_generator=self.schemas_generator
+            spec=spec,
+            output_dir=output_dir,
+            schemas_generator=self.schemas_generator,
+            asyncio=asyncio,
+        )
+        self.http_generator = HTTPGenerator(
+            spec=spec, output_dir=output_dir, asyncio=asyncio
         )
         self.spec = spec
         self.asyncio = asyncio
@@ -34,3 +42,4 @@ def generate(
         copy_tree(src=TEMPLATE_ROOT, dst=self.output_dir)
         self.schemas_generator.generate_schema_classes()
         self.clients_generator.generate_paths(api_url=url)
+        self.http_generator.generate_http_content()

src/generators/clients.py

@@ -23,12 +23,17 @@ class ClientsGenerator:
     schemas_generator: SchemasGenerator
 
     def __init__(
-        self, spec: Spec, output_dir: str, schemas_generator: SchemasGenerator
+        self,
+        spec: Spec,
+        output_dir: str,
+        schemas_generator: SchemasGenerator,
+        asyncio: bool,
     ) -> None:
         self.spec = spec
         self.output_dir = output_dir
         self.results = defaultdict(int)
         self.schemas_generator = schemas_generator
+        self.asyncio = asyncio
 
     def generate_paths(self, api_url: str) -> None:
         for path in self.spec["paths"].items():
@@ -49,6 +54,9 @@ def generate_function_args(self, parameters: List[Dict]) -> Dict[str, Any]:
         query_args = []
         path_args = []
         for p in parameters:
+            if p.get("$ref"):
+                # Not currently supporter
+                continue
             clean_key = clean_prop(p["name"])
             if clean_key in param_keys:
                 continue
@@ -78,16 +86,16 @@ def get_response_class_names(self, responses: Dict) -> List[str]:
         """
         response_classes = []
         for _, details in responses.items():
-            for _, content in details.get("content", {}).items():
+            for encoding, content in details.get("content", {}).items():
                 class_name = ""
                 if ref := content["schema"].get("$ref", False):
                     class_name = class_name_titled(
                         ref.replace("#/components/schemas/", "")
                     )
                 elif title := content["schema"].get("title", False):
-                    class_name = title
+                    class_name = class_name_titled(title)
                 else:
-                    raise "Cannot find a name for this class"
+                    class_name = class_name_titled(encoding)
                 response_classes.append(class_name)
         return list(set(response_classes))
 
@@ -97,7 +105,7 @@ def get_input_class_names(self, inputs: Dict) -> List[str]:
         """
         input_classes = []
         for _, details in inputs.items():
-            for _, content in details["content"].items():
+            for encoding, content in details.get("content", {}).items():
                 class_name = ""
                 if ref := content["schema"].get("$ref", False):
                     class_name = class_name_titled(
@@ -106,7 +114,8 @@ def get_input_class_names(self, inputs: Dict) -> List[str]:
                 elif title := content["schema"].get("title", False):
                     class_name = title
                 else:
-                    raise "Cannot find a name for this class"
+                    # No idea, using the encoding?
+                    class_name = encoding
                 class_name = class_name_titled(class_name)
                 input_classes.append(class_name)
         return list(set(input_classes))
@@ -115,6 +124,8 @@ def generate_response_types(self, responses: Dict) -> str:
         response_class_names = self.get_response_class_names(responses=responses)
         if len(response_class_names) > 1:
             return f"""typing.Union[{', '.join([f'schemas.{r}' for r in response_class_names])}]"""
+        elif len(response_class_names) == 0:
+            return "None"
         else:
             return f"schemas.{response_class_names[0]}"
 
@@ -126,6 +137,8 @@ def generate_input_types(self, request_body: Dict) -> str:
                 self.schemas_generator.generate_input_class(schema=request_body)
         if len(input_class_names) > 1:
             return f"""typing.Union[{', '.join([f'schemas.{r}' for r in input_class_names])}]"""
+        elif len(input_class_names) == 0:
+            return "None"
         else:
             return f"schemas.{input_class_names[0]}"
 
@@ -144,9 +157,9 @@ def generate_get_content(self, operation: Dict, api_url: str, path: str) -> None
         else:
             api_url = f"{self.parse_api_base_url(api_url)}{path}"
         CONTENT = f"""
-def {func_name}({function_arguments['return_string']}) -> {response_types}:
-    response = _get(f"{api_url}")
-    return _handle_response({func_name}, response)
+{self.asyncio and "async " or ""}def {func_name}({function_arguments['return_string']}) -> {response_types}:
+    response = {self.asyncio and "await " or ""}http.get(f"{api_url}")
+    return http.handle_response({func_name}, response)
     """
         self.results["get_methods"] += 1
         write_to_client(content=CONTENT, output_dir=self.output_dir)
@@ -155,16 +168,21 @@ def generate_post_content(self, operation: Dict, api_url: str, path: str) -> Non
         api_url = f"{self.parse_api_base_url(api_url)}{path}"
         response_types = self.generate_response_types(operation["responses"])
         func_name = get_func_name(operation, path)
-        input_class_name = self.generate_input_types({"": operation["requestBody"]})
+        if not operation.get("requestBody"):
+            input_class_name = "None"
+        else:
+            input_class_name = self.generate_input_types(
+                {"": operation.get("requestBody")}
+            )
         function_arguments = self.generate_function_args(
             operation.get("parameters", [])
         )
         FUNCTION_ARGS = f"""
 {function_arguments['return_string']}{function_arguments['return_string'] and ", "}data: {input_class_name}"""
         CONTENT = f"""
-def {func_name}({FUNCTION_ARGS}) -> {response_types}:
-    response = _post(f"{api_url}", data=data.model_dump())
-    return _handle_response({func_name}, response)
+{self.asyncio and "async " or ""}def {func_name}({FUNCTION_ARGS}) -> {response_types}:
+    response = {self.asyncio and "await " or ""}http.post(f"{api_url}", data=data and data.model_dump())
+    return http.handle_response({func_name}, response)
     """
         self.results["post_methods"] += 1
         write_to_client(content=CONTENT, output_dir=self.output_dir)

src/generators/http.py

@@ -0,0 +1,106 @@
+from collections import defaultdict
+from typing import Dict
+
+from openapi_core import Spec
+from rich.console import Console
+
+from src.writer import write_to_http
+
+console = Console()
+
+BEARER_CLIENT = """
+AUTH_TOKEN = environ.get("{bearer_token_key}")
+headers = dict(Authorization=f'Bearer {{AUTH_TOKEN}}')
+client = httpx.{client_type}(headers=headers)
+"""
+
+BASIC_CLIENT = """
+AUTH_USER = environ.get("{user_key}")
+AUTH_PASS = environ.get("{pass_key}")
+client = httpx.{client_type}(auth=(AUTH_USER, AUTH_PASS))
+"""
+
+NO_AUTH_CLIENT = """
+client = httpx.{client_type}()
+"""
+
+SYNC_METHODS = """
+def get(url: str) -> httpx.Response:
+    return client.get(url)
+
+
+def post(url: str, data: typing.Dict) -> httpx.Response:
+    return client.post(url, json=data)
+"""
+
+ASYNC_METHODS = """
+async def get(url: str) -> httpx.Response:
+    return await client.get(url)
+
+
+async def post(url: str, data: typing.Dict) -> httpx.Response:
+    return await client.post(url, json=data)
+"""
+
+
+def env_var(output_dir: str, key: str) -> str:
+    output_dir = output_dir.replace("/", "")
+    return f"{output_dir.upper()}_{key.upper()}"
+
+
+class HTTPGenerator:
+    """
+    Handles all the content generated in the clients.py file.
+    """
+
+    def __init__(self, spec: Spec, output_dir: str, asyncio: bool) -> None:
+        self.spec = spec
+        self.output_dir = output_dir
+        self.results: Dict[str, int] = defaultdict(int)
+        self.asyncio = asyncio
+
+    def generate_http_content(self) -> None:
+        client_generated = False
+        client_type = "AsyncClient" if self.asyncio else "Client"
+        if security_schemes := self.spec["components"].get("securitySchemes"):
+            console.log("Generating client with authentication...")
+            for _, info in security_schemes.items():
+                if (
+                    info["type"] == "http"
+                    and info["scheme"] in ["basic", "bearer"]
+                    and client_generated is False
+                ):
+                    client_generated = True
+                    if info["scheme"] == "bearer":
+                        test_key = env_var(output_dir=self.output_dir, key="AUTH_KEY")
+                        content = BEARER_CLIENT.format(
+                            client_type=client_type,
+                            bearer_token_key=f"{test_key}",
+                        )
+                        console.log(
+                            f"[yellow]Please use \n* {test_key}\nenvironment variable to use bearer authentication"
+                        )
+                    else:  # Can only be "basic" at this point
+                        user_key = env_var(
+                            output_dir=self.output_dir, key="AUTH_USER_KEY"
+                        )
+                        pass_key = env_var(
+                            output_dir=self.output_dir, key="AUTH_PASS_KEY"
+                        )
+                        console.log(
+                            f"[yellow]Please set \n* {user_key}\n* {pass_key} \nenvironment variable to use basic authentication"  # noqa
+                        )
+                        content = BASIC_CLIENT.format(
+                            client_type=client_type,
+                            user_key=f"{user_key}",
+                            pass_key=f"{pass_key}",
+                        )
+        if client_generated is False:
+            console.log(f"Generating {'async' if self.asyncio else 'sync'} client...")
+            content = NO_AUTH_CLIENT.format(client_type=client_type)
+            client_generated = True
+        write_to_http(content, output_dir=self.output_dir)
+        if self.asyncio:
+            write_to_http(ASYNC_METHODS, output_dir=self.output_dir)
+        else:
+            write_to_http(SYNC_METHODS, output_dir=self.output_dir)

src/generators/schemas.py

@@ -4,7 +4,7 @@
 from rich.console import Console
 
 from src.utils import class_name_titled, clean_prop, get_type
-from src.writer import write_to_response
+from src.writer import write_to_schemas
 
 console = Console()
 
@@ -55,24 +55,25 @@ def generate_class_properties(self, properties: Dict) -> str:
     def generate_input_class(self, schema: Dict) -> None:
         for _, schema_details in schema.items():
             content = schema_details["content"]
-            for _, input_schema in content.items():
+            for encoding, input_schema in content.items():
                 class_name = ""
                 if ref := input_schema["schema"].get("$ref", False):
                     class_name = class_name_titled(
                         ref.replace("#/components/schemas/", "")
                     )
                 elif title := input_schema["schema"].get("title", False):
-                    class_name = title
+                    class_name = class_name_titled(title)
                 else:
-                    raise "Cannot find a name for this class"
+                    # No idea, using the encoding?
+                    class_name = class_name_titled(encoding)
                 properties = self.generate_class_properties(
-                    input_schema["schema"]["properties"]
+                    input_schema["schema"].get("properties", {})
                 )
                 content = f"""
 class {class_name}(BaseModel):
 {properties if properties else "    pass"}
     """
-            write_to_response(
+            write_to_schemas(
                 content,
                 output_dir=self.output_dir,
             )
@@ -98,7 +99,7 @@ def generate_schema_classes(self) -> None:
 class {schema_key}({"Enum" if enum else "BaseModel"}):
 {properties if properties else "    pass"}
     """
-            write_to_response(
+            write_to_schemas(
                 content,
                 output_dir=self.output_dir,
             )

src/template/client.py

@@ -1,3 +1,3 @@
 import typing  # noqa
 from . import schemas  # noqa
-from .http import _handle_response, _get, _post  # noqa
+from . import http  # noqa

src/template/http.py

@@ -1,20 +1,11 @@
 import typing
+from os import environ  # noqa
 
-from httpx import Client, Response
+import httpx  # noqa
 from pydantic import ValidationError
 
-client = Client()
 
-
-def _get(url: str) -> Response:
-    return client.get(url)
-
-
-def _post(url: str, data: typing.Dict) -> Response:
-    return client.post(url, json=data)
-
-
-def _handle_response(func, response):
+def handle_response(func, response):
     """
     returns a response that matches the data neatly for a function
     """

src/utils.py

@@ -18,7 +18,7 @@ def class_name_titled(input_str: str) -> str:
     """
     # Capitalize the first letter always
     input_str = input_str[:1].title() + input_str[1:]
-    for badstr in [".", "-", "_", ">", "<"]:
+    for badstr in [".", "-", "_", ">", "<", "/"]:
         input_str = input_str.replace(badstr, " ")
     if " " in input_str:
         # Capitalize all the spaces
@@ -43,7 +43,7 @@ def clean_prop(input_str: str) -> str:
 
 def get_func_name(operation: Dict, path: str) -> str:
     if operation.get("operationId"):
-        return operation["operationId"].split("__")[0]
+        return class_name_titled(operation["operationId"].split("__")[0])
     # Probably 3.0.1
     return path.replace("/", "_").replace("-", "_")[1:]