From a28ab0a8a29379ce6b63820a47dd4a96828573c5 Mon Sep 17 00:00:00 2001 From: collerek Date: Tue, 18 May 2021 16:16:12 +0200 Subject: [PATCH] fixes for #199 and unreported choices bug --- docs/releases.md | 3 + ormar/fields/model_fields.py | 114 +++++++++++++++-------- ormar/models/helpers/validation.py | 53 +++++++++-- tests/test_fastapi/test_binary_fields.py | 13 ++- 4 files changed, 134 insertions(+), 49 deletions(-) diff --git a/docs/releases.md b/docs/releases.md index ba34c2f..d45dd44 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -3,6 +3,9 @@ ## 🐛 Fixes * Fix populating default values in pk_only child models [#202](https://github.com/collerek/ormar/issues/202) +* Fix mypy for LargeBinary fields with base64 str representation [#199](https://github.com/collerek/ormar/issues/199) +* Fix OpenAPI schema format for LargeBinary fields with base64 str representation [#199](https://github.com/collerek/ormar/issues/199) +* Fix OpenAPI choices encoding for LargeBinary fields with base64 str representation # 0.10.7 diff --git a/ormar/fields/model_fields.py b/ormar/fields/model_fields.py index 929903e..30628de 100644 --- a/ormar/fields/model_fields.py +++ b/ormar/fields/model_fields.py @@ -1,7 +1,7 @@ import datetime import decimal import uuid -from typing import Any, Optional, TYPE_CHECKING +from typing import Any, List, Literal, Optional, TYPE_CHECKING, Union, overload import pydantic import sqlalchemy @@ -426,52 +426,84 @@ class JSON(ModelFieldFactory, pydantic.Json): return sqlalchemy.JSON() -class LargeBinary(ModelFieldFactory, bytes): - """ - LargeBinary field factory that construct Field classes and populated their values. - """ +if TYPE_CHECKING: # pragma: nocover - _type = bytes - _sample = "bytes" + @overload + def LargeBinary( + max_length: int, *, represent_as_base64_str: Literal[True], **kwargs + ) -> str: + ... - def __new__( # type: ignore # noqa CFQ002 - cls, *, max_length: int, represent_as_base64_str: bool = False, **kwargs: Any - ) -> BaseField: # type: ignore - kwargs = { - **kwargs, - **{ - k: v - for k, v in locals().items() - if k not in ["cls", "__class__", "kwargs"] - }, - } - return super().__new__(cls, **kwargs) + @overload + def LargeBinary( + max_length: int, *, represent_as_base64_str: Literal[False], **kwargs + ) -> bytes: + ... - @classmethod - def get_column_type(cls, **kwargs: Any) -> Any: + @overload + def LargeBinary( + max_length: int, represent_as_base64_str: Literal[False] = ..., **kwargs + ) -> bytes: + ... + + def LargeBinary( + max_length: int, represent_as_base64_str: bool = False, **kwargs: Any + ) -> Union[str, bytes]: + pass + + +else: + + class LargeBinary(ModelFieldFactory, bytes): + """ + LargeBinary field factory that construct Field classes and populated their values. """ - Return proper type of db column for given field type. - Accepts required and optional parameters that each column type accepts. - :param kwargs: key, value pairs of sqlalchemy options - :type kwargs: Any - :return: initialized column with proper options - :rtype: sqlalchemy Column - """ - return sqlalchemy.LargeBinary(length=kwargs.get("max_length")) + _type = bytes + _sample = "bytes" - @classmethod - def validate(cls, **kwargs: Any) -> None: - """ - Used to validate if all required parameters on a given field type are set. - :param kwargs: all params passed during construction - :type kwargs: Any - """ - max_length = kwargs.get("max_length", None) - if max_length <= 0: - raise ModelDefinitionError( - "Parameter max_length is required for field LargeBinary" - ) + def __new__( # type: ignore # noqa CFQ002 + cls, + *, + max_length: int, + represent_as_base64_str: bool = False, + **kwargs: Any + ) -> BaseField: # type: ignore + kwargs = { + **kwargs, + **{ + k: v + for k, v in locals().items() + if k not in ["cls", "__class__", "kwargs"] + }, + } + return super().__new__(cls, **kwargs) + + @classmethod + def get_column_type(cls, **kwargs: Any) -> Any: + """ + Return proper type of db column for given field type. + Accepts required and optional parameters that each column type accepts. + + :param kwargs: key, value pairs of sqlalchemy options + :type kwargs: Any + :return: initialized column with proper options + :rtype: sqlalchemy Column + """ + return sqlalchemy.LargeBinary(length=kwargs.get("max_length")) + + @classmethod + def validate(cls, **kwargs: Any) -> None: + """ + Used to validate if all required parameters on a given field type are set. + :param kwargs: all params passed during construction + :type kwargs: Any + """ + max_length = kwargs.get("max_length", None) + if max_length <= 0: + raise ModelDefinitionError( + "Parameter max_length is required for field LargeBinary" + ) class BigInteger(Integer, int): diff --git a/ormar/models/helpers/validation.py b/ormar/models/helpers/validation.py index b972037..981517f 100644 --- a/ormar/models/helpers/validation.py +++ b/ormar/models/helpers/validation.py @@ -142,7 +142,10 @@ def generate_model_example(model: Type["Model"], relation_map: Dict = None) -> D ) for name, field in model.Meta.model_fields.items(): if not field.is_relation: - example[name] = field.__sample__ + if field.__type__ == bytes and field.represent_as_base64_str: + example[name] = "string" + else: + example[name] = field.__sample__ elif isinstance(relation_map, dict) and name in relation_map: example[name] = get_nested_model_example( name=name, field=field, relation_map=relation_map @@ -217,6 +220,44 @@ def get_pydantic_example_repr(type_: Any) -> Any: return "string" +def overwrite_example_and_description( + schema: Dict[str, Any], model: Type["Model"] +) -> None: + """ + Overwrites the example with properly nested children models. + Overwrites the description if it's taken from ormar.Model. + + :param schema: schema of current model + :type schema: Dict[str, Any] + :param model: model class + :type model: Type["Model"] + """ + schema["example"] = generate_model_example(model=model) + if "Main base class of ormar Model." in schema.get("description", ""): + schema["description"] = f"{model.__name__}" + + +def overwrite_binary_format(schema: Dict[str, Any], model: Type["Model"]) -> None: + """ + Overwrites format of the field if it's a LargeBinary field with + a flag to represent the field as base64 encoded string. + + :param schema: schema of current model + :type schema: Dict[str, Any] + :param model: model class + :type model: Type["Model"] + """ + for field_id, prop in schema.get("properties", {}).items(): + if ( + field_id in model._bytes_fields + and model.Meta.model_fields[field_id].represent_as_base64_str + ): + prop["format"] = "base64" + prop["enum"] = [ + base64.b64encode(choice).decode() for choice in prop["enum"] + ] + + def construct_modify_schema_function(fields_with_choices: List) -> SchemaExtraCallable: """ Modifies the schema to include fields with choices validator. @@ -237,9 +278,8 @@ def construct_modify_schema_function(fields_with_choices: List) -> SchemaExtraCa if field_id in fields_with_choices: prop["enum"] = list(model.Meta.model_fields[field_id].choices) prop["description"] = prop.get("description", "") + "An enumeration." - schema["example"] = generate_model_example(model=model) - if "Main base class of ormar Model." in schema.get("description", ""): - schema["description"] = f"{model.__name__}" + overwrite_example_and_description(schema=schema, model=model) + overwrite_binary_format(schema=schema, model=model) return staticmethod(schema_extra) # type: ignore @@ -256,9 +296,8 @@ def construct_schema_function_without_choices() -> SchemaExtraCallable: """ def schema_extra(schema: Dict[str, Any], model: Type["Model"]) -> None: - schema["example"] = generate_model_example(model=model) - if "Main base class of ormar Model." in schema.get("description", ""): - schema["description"] = f"{model.__name__}" + overwrite_example_and_description(schema=schema, model=model) + overwrite_binary_format(schema=schema, model=model) return staticmethod(schema_extra) # type: ignore diff --git a/tests/test_fastapi/test_binary_fields.py b/tests/test_fastapi/test_binary_fields.py index af38e18..82f701f 100644 --- a/tests/test_fastapi/test_binary_fields.py +++ b/tests/test_fastapi/test_binary_fields.py @@ -52,7 +52,7 @@ class BinaryThing(ormar.Model): id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4) name: str = ormar.Text(default="") - bt: bytes = ormar.LargeBinary( + bt: str = ormar.LargeBinary( max_length=1000, choices=[blob3, blob4, blob5, blob6], represent_as_base64_str=True, @@ -89,3 +89,14 @@ def test_read_main(): assert response.json()[0]["bt"] == base64.b64encode(blob3).decode() thing = BinaryThing(**response.json()[0]) assert thing.__dict__["bt"] == blob3 + + +def test_schema(): + schema = BinaryThing.schema() + assert schema["properties"]["bt"]["format"] == "base64" + converted_choices = ["7g==", "/w==", "8CiMKA==", "wyg="] + assert len(schema["properties"]["bt"]["enum"]) == 4 + assert all( + choice in schema["properties"]["bt"]["enum"] for choice in converted_choices + ) + assert schema["example"]["bt"] == "string"