Flask

Basic

from dataclasses import dataclass
from typing import Annotated, Optional, Tuple

from flask import Flask, jsonify, request
from flask.typing import ResponseValue

from koda_validate import StringValidator, DataclassValidator, EmailPredicate
from koda_validate.serialization import to_serializable_errs

app = Flask(__name__)


@dataclass
class ContactForm:
    name: str
    message: str
    # `Annotated` `Validator`s are used if found
    email: Annotated[str, StringValidator(EmailPredicate())]
    subject: Optional[str] = None


@app.route("/contact", methods=["POST"])
def contact_api() -> Tuple[ResponseValue, int]:
    result = DataclassValidator(ContactForm)(request.json)
    match result:
        case Valid(contact_form):
            print(contact_form)  # do something with the valid data
            return {"success": True}, 200
        case Invalid() as inv:
            return jsonify(to_serializable_errs(inv)), 400


if __name__ == "__main__":
    app.run()

Note

In the example above, ContactForm is a dataclass, so we use a DataclassValidator. We could have used a TypedDict and TypedDictValidator, or a NamedTuple and NamedTupleValidator, and the code would have been essentially the same.

Fuller Example (with Async)

import asyncio
from typing import Annotated, Optional, Tuple, TypedDict

from flask import Flask, jsonify, request
from flask.typing import ResponseValue

from koda_validate import *
from koda_validate.serialization import SerializableErr, to_serializable_errs

app = Flask(__name__)


class Captcha(TypedDict):
    seed: Annotated[str, StringValidator(ExactLength(16))]
    response: Annotated[str, StringValidator(MaxLength(16))]


async def validate_captcha(captcha: Captcha) -> Optional[ErrType]:
    """
    after we validate that the seed and response on their own,
    we need to check our database to make sure the response is correct
    """

    async def pretend_check_captcha_service(seed: str, response: str) -> bool:
        await asyncio.sleep(0.01)  # pretend to call
        return seed == response[::-1]

    if await pretend_check_captcha_service(captcha["seed"], captcha["response"]):
        # everything's valid
        return None
    else:
        return SerializableErr({"response": "bad captcha response"})


class ContactForm(TypedDict):
    email: Annotated[str, StringValidator(EmailPredicate())]
    message: Annotated[str, StringValidator(MaxLength(500), MinLength(10))]
    captcha: Annotated[
        Captcha,
        # explicitly adding some extra validation
        TypedDictValidator(Captcha, validate_object_async=validate_captcha),
    ]


contact_validator = TypedDictValidator(ContactForm)


@app.route("/contact", methods=["POST"])
async def contact_api() -> Tuple[ResponseValue, int]:
    result = await contact_validator.validate_async(request.json)
    match result:
        case Valid(contact_form):
            print(contact_form)
            return {"success": True}, 200
        case Invalid() as inv:
            return jsonify(to_serializable_errs(inv)), 400


# if you want a JSON Schema from a ``Validator``, there's `to_json_schema()`
# schema = to_json_schema(contact_validator)
# hook_into_some_api_definition(schema)


if __name__ == "__main__":
    app.run()