Fix python-fastapi signature of parameters for method (#19830)

* Fix python-fastapi signature of parameters for method

* update

---------

Co-authored-by: Diorcet Yann <diorcet.yann@gmail.com>
This commit is contained in:
William Cheng 2024-10-10 15:36:22 +08:00 committed by GitHub
parent 45fa4384e7
commit b357744048
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 183 additions and 55 deletions

View File

@ -50,6 +50,7 @@ import org.openapitools.codegen.CodegenModel;
import org.openapitools.codegen.CodegenOperation; import org.openapitools.codegen.CodegenOperation;
import org.openapitools.codegen.CodegenParameter; import org.openapitools.codegen.CodegenParameter;
import org.openapitools.codegen.CodegenProperty; import org.openapitools.codegen.CodegenProperty;
import org.openapitools.codegen.CodegenResponse;
import org.openapitools.codegen.DefaultCodegen; import org.openapitools.codegen.DefaultCodegen;
import org.openapitools.codegen.GeneratorLanguage; import org.openapitools.codegen.GeneratorLanguage;
import org.openapitools.codegen.IJsonSchemaValidationProperties; import org.openapitools.codegen.IJsonSchemaValidationProperties;
@ -1264,6 +1265,23 @@ public abstract class AbstractPythonCodegen extends DefaultCodegen implements Co
); );
} }
// update typing import for operation responses type
// only python-fastapi needs this at the moment
if (this instanceof PythonFastAPIServerCodegen) {
for (CodegenResponse response : operation.responses) {
// Not interested in the result, only in the update of the imports
getPydanticType(
response.returnProperty,
modelImports,
exampleImports,
postponedModelImports,
postponedExampleImports,
moduleImports,
null
);
}
}
// add import for code samples // add import for code samples
// import models one by one // import models one by one
if (!exampleImports.isEmpty()) { if (!exampleImports.isEmpty()) {

View File

@ -219,6 +219,8 @@ public class PythonFastAPIServerCodegen extends AbstractPythonCodegen {
@Override @Override
public OperationsMap postProcessOperationsWithModels(OperationsMap objs, List<ModelMap> allModels) { public OperationsMap postProcessOperationsWithModels(OperationsMap objs, List<ModelMap> allModels) {
super.postProcessOperationsWithModels(objs, allModels);
OperationMap operations = objs.getOperations(); OperationMap operations = objs.getOperations();
// Set will make sure that no duplicated items are used. // Set will make sure that no duplicated items are used.
Set<String> securityImports = new HashSet<>(); Set<String> securityImports = new HashSet<>();

View File

@ -1 +1 @@
{{#isString}}str{{/isString}}{{#isInteger}}int{{/isInteger}}{{#isLong}}int{{/isLong}}{{#isFloat}}float{{/isFloat}}{{#isDouble}}float{{/isDouble}}{{#isByteArray}}str{{/isByteArray}}{{#isBinary}}str{{/isBinary}}{{#isBoolean}}bool{{/isBoolean}}{{#isDate}}str{{/isDate}}{{#isDateTime}}str{{/isDateTime}}{{#isModel}}{{dataType}}{{/isModel}}{{#isContainer}}{{dataType}}{{/isContainer}} {{{vendorExtensions.x-py-typing}}}

View File

@ -0,0 +1,60 @@
package org.openapitools.codegen.python;
import com.google.common.collect.Sets;
import io.swagger.parser.OpenAPIParser;
import io.swagger.v3.oas.models.OpenAPI;
import io.swagger.v3.oas.models.Operation;
import io.swagger.v3.oas.models.media.ArraySchema;
import io.swagger.v3.oas.models.media.Schema;
import io.swagger.v3.parser.core.models.ParseOptions;
import org.openapitools.codegen.*;
import org.openapitools.codegen.languages.PythonClientCodegen;
import org.openapitools.codegen.languages.PythonFastAPIServerCodegen;
import org.openapitools.codegen.languages.features.CXFServerFeatures;
import org.testng.Assert;
import org.testng.annotations.Test;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import static org.openapitools.codegen.TestUtils.assertFileContains;
import static org.openapitools.codegen.TestUtils.assertFileExists;
public class PythonFastAPIServerCodegenTest {
// Helper function, intended to reduce boilerplate
static private String generateFiles(DefaultCodegen codegen, String filePath) throws IOException {
final File output = Files.createTempDirectory("test").toFile().getCanonicalFile();
output.deleteOnExit();
final String outputPath = output.getAbsolutePath().replace('\\', '/');
codegen.setOutputDir(output.getAbsolutePath());
codegen.additionalProperties().put(CXFServerFeatures.LOAD_TEST_DATA_FROM_FILE, "true");
final ClientOptInput input = new ClientOptInput();
final OpenAPI openAPI = new OpenAPIParser().readLocation(filePath, null, new ParseOptions()).getOpenAPI();
input.openAPI(openAPI);
input.config(codegen);
final DefaultGenerator generator = new DefaultGenerator();
final List<File> files = generator.opts(input).generate();
Assert.assertTrue(files.size() > 0);
return outputPath + "/";
}
@Test(description = "test containerType in parameters")
public void testContainerType() throws IOException {
final DefaultCodegen codegen = new PythonFastAPIServerCodegen();
final String outputPath = generateFiles(codegen, "src/test/resources/bugs/pr_18691.json");
final Path p = Paths.get(outputPath + "src/openapi_server/apis/default_api.py");
assertFileExists(p);
assertFileContains(p, "body: Optional[Dict[str, Any]] = Body(None, description=\"\"),");
}
}

View File

@ -0,0 +1,24 @@
{
"openapi": "3.0.1",
"info": {
"title": "OpenAPI definition",
"version": "v0"
},
"paths": {
"/licensing/token/renew": {
"post": {
"description": "Manually ask license issuer for a new token. Available in Grafana Enterprise v7.4+.\n\nYou need to have a permission with action `licensing:update`.",
"operationId": "postRenewLicenseToken",
"requestBody": {
"content": {
"application/json": {
"schema": {
"type": "object"
}
}
}
}
}
}
}
}

View File

@ -23,6 +23,9 @@ from fastapi import ( # noqa: F401
) )
from openapi_server.models.extra_models import TokenModel # noqa: F401 from openapi_server.models.extra_models import TokenModel # noqa: F401
from pydantic import Field, StrictStr
from typing import Any, Optional
from typing_extensions import Annotated
router = APIRouter() router = APIRouter()
@ -43,8 +46,8 @@ for _, name, _ in pkgutil.iter_modules(ns_pkg.__path__, ns_pkg.__name__ + "."):
response_model_by_alias=True, response_model_by_alias=True,
) )
async def fake_query_param_default( async def fake_query_param_default(
has_default: str = Query('Hello World', description="has default value", alias="hasDefault"), has_default: Annotated[Optional[StrictStr], Field(description="has default value")] = Query('Hello World', description="has default value", alias="hasDefault"),
no_default: str = Query(None, description="no default value", alias="noDefault"), no_default: Annotated[Optional[StrictStr], Field(description="no default value")] = Query(None, description="no default value", alias="noDefault"),
) -> None: ) -> None:
"""""" """"""
if not BaseFakeApi.subclasses: if not BaseFakeApi.subclasses:

View File

@ -2,6 +2,9 @@
from typing import ClassVar, Dict, List, Tuple # noqa: F401 from typing import ClassVar, Dict, List, Tuple # noqa: F401
from pydantic import Field, StrictStr
from typing import Any, Optional
from typing_extensions import Annotated
class BaseFakeApi: class BaseFakeApi:
@ -12,8 +15,8 @@ class BaseFakeApi:
BaseFakeApi.subclasses = BaseFakeApi.subclasses + (cls,) BaseFakeApi.subclasses = BaseFakeApi.subclasses + (cls,)
async def fake_query_param_default( async def fake_query_param_default(
self, self,
has_default: str, has_default: Annotated[Optional[StrictStr], Field(description="has default value")],
no_default: str, no_default: Annotated[Optional[StrictStr], Field(description="no default value")],
) -> None: ) -> None:
"""""" """"""
... ...

View File

@ -23,6 +23,9 @@ from fastapi import ( # noqa: F401
) )
from openapi_server.models.extra_models import TokenModel # noqa: F401 from openapi_server.models.extra_models import TokenModel # noqa: F401
from pydantic import Field, StrictBytes, StrictInt, StrictStr, field_validator
from typing import Any, List, Optional, Tuple, Union
from typing_extensions import Annotated
from openapi_server.models.api_response import ApiResponse from openapi_server.models.api_response import ApiResponse
from openapi_server.models.pet import Pet from openapi_server.models.pet import Pet
from openapi_server.security_api import get_token_petstore_auth, get_token_api_key from openapi_server.security_api import get_token_petstore_auth, get_token_api_key
@ -45,7 +48,7 @@ for _, name, _ in pkgutil.iter_modules(ns_pkg.__path__, ns_pkg.__name__ + "."):
response_model_by_alias=True, response_model_by_alias=True,
) )
async def add_pet( async def add_pet(
pet: Pet = Body(None, description="Pet object that needs to be added to the store"), pet: Annotated[Pet, Field(description="Pet object that needs to be added to the store")] = Body(None, description="Pet object that needs to be added to the store"),
token_petstore_auth: TokenModel = Security( token_petstore_auth: TokenModel = Security(
get_token_petstore_auth, scopes=["write:pets", "read:pets"] get_token_petstore_auth, scopes=["write:pets", "read:pets"]
), ),
@ -66,8 +69,8 @@ async def add_pet(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def delete_pet( async def delete_pet(
petId: int = Path(..., description="Pet id to delete"), petId: Annotated[StrictInt, Field(description="Pet id to delete")] = Path(..., description="Pet id to delete"),
api_key: str = Header(None, description=""), api_key: Optional[StrictStr] = Header(None, description=""),
token_petstore_auth: TokenModel = Security( token_petstore_auth: TokenModel = Security(
get_token_petstore_auth, scopes=["write:pets", "read:pets"] get_token_petstore_auth, scopes=["write:pets", "read:pets"]
), ),
@ -89,7 +92,7 @@ async def delete_pet(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def find_pets_by_status( async def find_pets_by_status(
status: List[str] = Query(None, description="Status values that need to be considered for filter", alias="status"), status: Annotated[List[StrictStr], Field(description="Status values that need to be considered for filter")] = Query(None, description="Status values that need to be considered for filter", alias="status"),
token_petstore_auth: TokenModel = Security( token_petstore_auth: TokenModel = Security(
get_token_petstore_auth, scopes=["read:pets"] get_token_petstore_auth, scopes=["read:pets"]
), ),
@ -111,7 +114,7 @@ async def find_pets_by_status(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def find_pets_by_tags( async def find_pets_by_tags(
tags: List[str] = Query(None, description="Tags to filter by", alias="tags"), tags: Annotated[List[StrictStr], Field(description="Tags to filter by")] = Query(None, description="Tags to filter by", alias="tags"),
token_petstore_auth: TokenModel = Security( token_petstore_auth: TokenModel = Security(
get_token_petstore_auth, scopes=["read:pets"] get_token_petstore_auth, scopes=["read:pets"]
), ),
@ -134,7 +137,7 @@ async def find_pets_by_tags(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def get_pet_by_id( async def get_pet_by_id(
petId: int = Path(..., description="ID of pet to return"), petId: Annotated[StrictInt, Field(description="ID of pet to return")] = Path(..., description="ID of pet to return"),
token_api_key: TokenModel = Security( token_api_key: TokenModel = Security(
get_token_api_key get_token_api_key
), ),
@ -158,7 +161,7 @@ async def get_pet_by_id(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def update_pet( async def update_pet(
pet: Pet = Body(None, description="Pet object that needs to be added to the store"), pet: Annotated[Pet, Field(description="Pet object that needs to be added to the store")] = Body(None, description="Pet object that needs to be added to the store"),
token_petstore_auth: TokenModel = Security( token_petstore_auth: TokenModel = Security(
get_token_petstore_auth, scopes=["write:pets", "read:pets"] get_token_petstore_auth, scopes=["write:pets", "read:pets"]
), ),
@ -179,9 +182,9 @@ async def update_pet(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def update_pet_with_form( async def update_pet_with_form(
petId: int = Path(..., description="ID of pet that needs to be updated"), petId: Annotated[StrictInt, Field(description="ID of pet that needs to be updated")] = Path(..., description="ID of pet that needs to be updated"),
name: str = Form(None, description="Updated name of the pet"), name: Annotated[Optional[StrictStr], Field(description="Updated name of the pet")] = Form(None, description="Updated name of the pet"),
status: str = Form(None, description="Updated status of the pet"), status: Annotated[Optional[StrictStr], Field(description="Updated status of the pet")] = Form(None, description="Updated status of the pet"),
token_petstore_auth: TokenModel = Security( token_petstore_auth: TokenModel = Security(
get_token_petstore_auth, scopes=["write:pets", "read:pets"] get_token_petstore_auth, scopes=["write:pets", "read:pets"]
), ),
@ -202,9 +205,9 @@ async def update_pet_with_form(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def upload_file( async def upload_file(
petId: int = Path(..., description="ID of pet to update"), petId: Annotated[StrictInt, Field(description="ID of pet to update")] = Path(..., description="ID of pet to update"),
additional_metadata: str = Form(None, description="Additional data to pass to server"), additional_metadata: Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")] = Form(None, description="Additional data to pass to server"),
file: str = Form(None, description="file to upload"), file: Annotated[Optional[Union[StrictBytes, StrictStr, Tuple[StrictStr, StrictBytes]]], Field(description="file to upload")] = Form(None, description="file to upload"),
token_petstore_auth: TokenModel = Security( token_petstore_auth: TokenModel = Security(
get_token_petstore_auth, scopes=["write:pets", "read:pets"] get_token_petstore_auth, scopes=["write:pets", "read:pets"]
), ),

View File

@ -2,6 +2,9 @@
from typing import ClassVar, Dict, List, Tuple # noqa: F401 from typing import ClassVar, Dict, List, Tuple # noqa: F401
from pydantic import Field, StrictBytes, StrictInt, StrictStr, field_validator
from typing import Any, List, Optional, Tuple, Union
from typing_extensions import Annotated
from openapi_server.models.api_response import ApiResponse from openapi_server.models.api_response import ApiResponse
from openapi_server.models.pet import Pet from openapi_server.models.pet import Pet
from openapi_server.security_api import get_token_petstore_auth, get_token_api_key from openapi_server.security_api import get_token_petstore_auth, get_token_api_key
@ -14,7 +17,7 @@ class BasePetApi:
BasePetApi.subclasses = BasePetApi.subclasses + (cls,) BasePetApi.subclasses = BasePetApi.subclasses + (cls,)
async def add_pet( async def add_pet(
self, self,
pet: Pet, pet: Annotated[Pet, Field(description="Pet object that needs to be added to the store")],
) -> Pet: ) -> Pet:
"""""" """"""
... ...
@ -22,8 +25,8 @@ class BasePetApi:
async def delete_pet( async def delete_pet(
self, self,
petId: int, petId: Annotated[StrictInt, Field(description="Pet id to delete")],
api_key: str, api_key: Optional[StrictStr],
) -> None: ) -> None:
"""""" """"""
... ...
@ -31,7 +34,7 @@ class BasePetApi:
async def find_pets_by_status( async def find_pets_by_status(
self, self,
status: List[str], status: Annotated[List[StrictStr], Field(description="Status values that need to be considered for filter")],
) -> List[Pet]: ) -> List[Pet]:
"""Multiple status values can be provided with comma separated strings""" """Multiple status values can be provided with comma separated strings"""
... ...
@ -39,7 +42,7 @@ class BasePetApi:
async def find_pets_by_tags( async def find_pets_by_tags(
self, self,
tags: List[str], tags: Annotated[List[StrictStr], Field(description="Tags to filter by")],
) -> List[Pet]: ) -> List[Pet]:
"""Multiple tags can be provided with comma separated strings. Use tag1, tag2, tag3 for testing.""" """Multiple tags can be provided with comma separated strings. Use tag1, tag2, tag3 for testing."""
... ...
@ -47,7 +50,7 @@ class BasePetApi:
async def get_pet_by_id( async def get_pet_by_id(
self, self,
petId: int, petId: Annotated[StrictInt, Field(description="ID of pet to return")],
) -> Pet: ) -> Pet:
"""Returns a single pet""" """Returns a single pet"""
... ...
@ -55,7 +58,7 @@ class BasePetApi:
async def update_pet( async def update_pet(
self, self,
pet: Pet, pet: Annotated[Pet, Field(description="Pet object that needs to be added to the store")],
) -> Pet: ) -> Pet:
"""""" """"""
... ...
@ -63,9 +66,9 @@ class BasePetApi:
async def update_pet_with_form( async def update_pet_with_form(
self, self,
petId: int, petId: Annotated[StrictInt, Field(description="ID of pet that needs to be updated")],
name: str, name: Annotated[Optional[StrictStr], Field(description="Updated name of the pet")],
status: str, status: Annotated[Optional[StrictStr], Field(description="Updated status of the pet")],
) -> None: ) -> None:
"""""" """"""
... ...
@ -73,9 +76,9 @@ class BasePetApi:
async def upload_file( async def upload_file(
self, self,
petId: int, petId: Annotated[StrictInt, Field(description="ID of pet to update")],
additional_metadata: str, additional_metadata: Annotated[Optional[StrictStr], Field(description="Additional data to pass to server")],
file: str, file: Annotated[Optional[Union[StrictBytes, StrictStr, Tuple[StrictStr, StrictBytes]]], Field(description="file to upload")],
) -> ApiResponse: ) -> ApiResponse:
"""""" """"""
... ...

View File

@ -23,6 +23,9 @@ from fastapi import ( # noqa: F401
) )
from openapi_server.models.extra_models import TokenModel # noqa: F401 from openapi_server.models.extra_models import TokenModel # noqa: F401
from pydantic import Field, StrictInt, StrictStr
from typing import Any, Dict
from typing_extensions import Annotated
from openapi_server.models.order import Order from openapi_server.models.order import Order
from openapi_server.security_api import get_token_api_key from openapi_server.security_api import get_token_api_key
@ -44,7 +47,7 @@ for _, name, _ in pkgutil.iter_modules(ns_pkg.__path__, ns_pkg.__name__ + "."):
response_model_by_alias=True, response_model_by_alias=True,
) )
async def delete_order( async def delete_order(
orderId: str = Path(..., description="ID of the order that needs to be deleted"), orderId: Annotated[StrictStr, Field(description="ID of the order that needs to be deleted")] = Path(..., description="ID of the order that needs to be deleted"),
) -> None: ) -> None:
"""For valid response try integer IDs with value &lt; 1000. Anything above 1000 or nonintegers will generate API errors""" """For valid response try integer IDs with value &lt; 1000. Anything above 1000 or nonintegers will generate API errors"""
if not BaseStoreApi.subclasses: if not BaseStoreApi.subclasses:
@ -84,7 +87,7 @@ async def get_inventory(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def get_order_by_id( async def get_order_by_id(
orderId: int = Path(..., description="ID of pet that needs to be fetched", ge=1, le=5), orderId: Annotated[int, Field(le=5, strict=True, ge=1, description="ID of pet that needs to be fetched")] = Path(..., description="ID of pet that needs to be fetched", ge=1, le=5),
) -> Order: ) -> Order:
"""For valid response try integer IDs with value &lt;&#x3D; 5 or &gt; 10. Other values will generate exceptions""" """For valid response try integer IDs with value &lt;&#x3D; 5 or &gt; 10. Other values will generate exceptions"""
if not BaseStoreApi.subclasses: if not BaseStoreApi.subclasses:
@ -103,7 +106,7 @@ async def get_order_by_id(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def place_order( async def place_order(
order: Order = Body(None, description="order placed for purchasing the pet"), order: Annotated[Order, Field(description="order placed for purchasing the pet")] = Body(None, description="order placed for purchasing the pet"),
) -> Order: ) -> Order:
"""""" """"""
if not BaseStoreApi.subclasses: if not BaseStoreApi.subclasses:

View File

@ -2,6 +2,9 @@
from typing import ClassVar, Dict, List, Tuple # noqa: F401 from typing import ClassVar, Dict, List, Tuple # noqa: F401
from pydantic import Field, StrictInt, StrictStr
from typing import Any, Dict
from typing_extensions import Annotated
from openapi_server.models.order import Order from openapi_server.models.order import Order
from openapi_server.security_api import get_token_api_key from openapi_server.security_api import get_token_api_key
@ -13,7 +16,7 @@ class BaseStoreApi:
BaseStoreApi.subclasses = BaseStoreApi.subclasses + (cls,) BaseStoreApi.subclasses = BaseStoreApi.subclasses + (cls,)
async def delete_order( async def delete_order(
self, self,
orderId: str, orderId: Annotated[StrictStr, Field(description="ID of the order that needs to be deleted")],
) -> None: ) -> None:
"""For valid response try integer IDs with value &lt; 1000. Anything above 1000 or nonintegers will generate API errors""" """For valid response try integer IDs with value &lt; 1000. Anything above 1000 or nonintegers will generate API errors"""
... ...
@ -28,7 +31,7 @@ class BaseStoreApi:
async def get_order_by_id( async def get_order_by_id(
self, self,
orderId: int, orderId: Annotated[int, Field(le=5, strict=True, ge=1, description="ID of pet that needs to be fetched")],
) -> Order: ) -> Order:
"""For valid response try integer IDs with value &lt;&#x3D; 5 or &gt; 10. Other values will generate exceptions""" """For valid response try integer IDs with value &lt;&#x3D; 5 or &gt; 10. Other values will generate exceptions"""
... ...
@ -36,7 +39,7 @@ class BaseStoreApi:
async def place_order( async def place_order(
self, self,
order: Order, order: Annotated[Order, Field(description="order placed for purchasing the pet")],
) -> Order: ) -> Order:
"""""" """"""
... ...

View File

@ -23,6 +23,9 @@ from fastapi import ( # noqa: F401
) )
from openapi_server.models.extra_models import TokenModel # noqa: F401 from openapi_server.models.extra_models import TokenModel # noqa: F401
from pydantic import Field, StrictStr, field_validator
from typing import Any, List
from typing_extensions import Annotated
from openapi_server.models.user import User from openapi_server.models.user import User
from openapi_server.security_api import get_token_api_key from openapi_server.security_api import get_token_api_key
@ -43,7 +46,7 @@ for _, name, _ in pkgutil.iter_modules(ns_pkg.__path__, ns_pkg.__name__ + "."):
response_model_by_alias=True, response_model_by_alias=True,
) )
async def create_user( async def create_user(
user: User = Body(None, description="Created user object"), user: Annotated[User, Field(description="Created user object")] = Body(None, description="Created user object"),
token_api_key: TokenModel = Security( token_api_key: TokenModel = Security(
get_token_api_key get_token_api_key
), ),
@ -64,7 +67,7 @@ async def create_user(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def create_users_with_array_input( async def create_users_with_array_input(
user: List[User] = Body(None, description="List of user object"), user: Annotated[List[User], Field(description="List of user object")] = Body(None, description="List of user object"),
token_api_key: TokenModel = Security( token_api_key: TokenModel = Security(
get_token_api_key get_token_api_key
), ),
@ -85,7 +88,7 @@ async def create_users_with_array_input(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def create_users_with_list_input( async def create_users_with_list_input(
user: List[User] = Body(None, description="List of user object"), user: Annotated[List[User], Field(description="List of user object")] = Body(None, description="List of user object"),
token_api_key: TokenModel = Security( token_api_key: TokenModel = Security(
get_token_api_key get_token_api_key
), ),
@ -107,7 +110,7 @@ async def create_users_with_list_input(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def delete_user( async def delete_user(
username: str = Path(..., description="The name that needs to be deleted"), username: Annotated[StrictStr, Field(description="The name that needs to be deleted")] = Path(..., description="The name that needs to be deleted"),
token_api_key: TokenModel = Security( token_api_key: TokenModel = Security(
get_token_api_key get_token_api_key
), ),
@ -130,7 +133,7 @@ async def delete_user(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def get_user_by_name( async def get_user_by_name(
username: str = Path(..., description="The name that needs to be fetched. Use user1 for testing."), username: Annotated[StrictStr, Field(description="The name that needs to be fetched. Use user1 for testing.")] = Path(..., description="The name that needs to be fetched. Use user1 for testing."),
) -> User: ) -> User:
"""""" """"""
if not BaseUserApi.subclasses: if not BaseUserApi.subclasses:
@ -149,8 +152,8 @@ async def get_user_by_name(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def login_user( async def login_user(
username: str = Query(None, description="The user name for login", alias="username", regex=r"/^[a-zA-Z0-9]+[a-zA-Z0-9\.\-_]*[a-zA-Z0-9]+$/"), username: Annotated[str, Field(strict=True, description="The user name for login")] = Query(None, description="The user name for login", alias="username", regex=r"/^[a-zA-Z0-9]+[a-zA-Z0-9\.\-_]*[a-zA-Z0-9]+$/"),
password: str = Query(None, description="The password for login in clear text", alias="password"), password: Annotated[StrictStr, Field(description="The password for login in clear text")] = Query(None, description="The password for login in clear text", alias="password"),
) -> str: ) -> str:
"""""" """"""
if not BaseUserApi.subclasses: if not BaseUserApi.subclasses:
@ -189,8 +192,8 @@ async def logout_user(
response_model_by_alias=True, response_model_by_alias=True,
) )
async def update_user( async def update_user(
username: str = Path(..., description="name that need to be deleted"), username: Annotated[StrictStr, Field(description="name that need to be deleted")] = Path(..., description="name that need to be deleted"),
user: User = Body(None, description="Updated user object"), user: Annotated[User, Field(description="Updated user object")] = Body(None, description="Updated user object"),
token_api_key: TokenModel = Security( token_api_key: TokenModel = Security(
get_token_api_key get_token_api_key
), ),

View File

@ -2,6 +2,9 @@
from typing import ClassVar, Dict, List, Tuple # noqa: F401 from typing import ClassVar, Dict, List, Tuple # noqa: F401
from pydantic import Field, StrictStr, field_validator
from typing import Any, List
from typing_extensions import Annotated
from openapi_server.models.user import User from openapi_server.models.user import User
from openapi_server.security_api import get_token_api_key from openapi_server.security_api import get_token_api_key
@ -13,7 +16,7 @@ class BaseUserApi:
BaseUserApi.subclasses = BaseUserApi.subclasses + (cls,) BaseUserApi.subclasses = BaseUserApi.subclasses + (cls,)
async def create_user( async def create_user(
self, self,
user: User, user: Annotated[User, Field(description="Created user object")],
) -> None: ) -> None:
"""This can only be done by the logged in user.""" """This can only be done by the logged in user."""
... ...
@ -21,7 +24,7 @@ class BaseUserApi:
async def create_users_with_array_input( async def create_users_with_array_input(
self, self,
user: List[User], user: Annotated[List[User], Field(description="List of user object")],
) -> None: ) -> None:
"""""" """"""
... ...
@ -29,7 +32,7 @@ class BaseUserApi:
async def create_users_with_list_input( async def create_users_with_list_input(
self, self,
user: List[User], user: Annotated[List[User], Field(description="List of user object")],
) -> None: ) -> None:
"""""" """"""
... ...
@ -37,7 +40,7 @@ class BaseUserApi:
async def delete_user( async def delete_user(
self, self,
username: str, username: Annotated[StrictStr, Field(description="The name that needs to be deleted")],
) -> None: ) -> None:
"""This can only be done by the logged in user.""" """This can only be done by the logged in user."""
... ...
@ -45,7 +48,7 @@ class BaseUserApi:
async def get_user_by_name( async def get_user_by_name(
self, self,
username: str, username: Annotated[StrictStr, Field(description="The name that needs to be fetched. Use user1 for testing.")],
) -> User: ) -> User:
"""""" """"""
... ...
@ -53,8 +56,8 @@ class BaseUserApi:
async def login_user( async def login_user(
self, self,
username: str, username: Annotated[str, Field(strict=True, description="The user name for login")],
password: str, password: Annotated[StrictStr, Field(description="The password for login in clear text")],
) -> str: ) -> str:
"""""" """"""
... ...
@ -69,8 +72,8 @@ class BaseUserApi:
async def update_user( async def update_user(
self, self,
username: str, username: Annotated[StrictStr, Field(description="name that need to be deleted")],
user: User, user: Annotated[User, Field(description="Updated user object")],
) -> None: ) -> None:
"""This can only be done by the logged in user.""" """This can only be done by the logged in user."""
... ...