Example of REST API models with filtering

event.json

{
  "body": "{\"sort\": {\"sortBy\": \"id\"}, \"filters\": {\"samples\": {\"concentration\": {\"min_value\": 1.2}, \"competitor\": \"abc\"}, \"hit_strength\": {\"min_value\": 20.0, \"max_value\": 50.0}}}",
  "headers": {
    "Accept": "*/*",
    "Content-Length": "26",
    "Content-Type": "application/x-www-form-urlencoded",
    "Host": "127.0.0.1:3000",
    "User-Agent": "curl/7.68.0",
    "X-Forwarded-Port": "3000",
    "X-Forwarded-Proto": "http"
  },
  "httpMethod": "POST",
  "isBase64Encoded": false,
  "multiValueHeaders": {
    "Accept": [
      "*/*"
    ],
    "Content-Length": [
      "26"
    ],
    "Content-Type": [
      "application/x-www-form-urlencoded"
    ],
    "Host": [
      "127.0.0.1:3000"
    ],
    "User-Agent": [
      "curl/7.68.0"
    ],
    "X-Forwarded-Port": [
      "3000"
    ],
    "X-Forwarded-Proto": [
      "http"
    ]
  },
  "multiValueQueryStringParameters": null,
  "path": "/phagehits",
  "pathParameters": null,
  "queryStringParameters": null,
  "requestContext": {
    "accountId": "123456789012",
    "apiId": "1234567890",
    "domainName": "127.0.0.1:3000",
    "extendedRequestId": null,
    "httpMethod": "POST",
    "identity": {
      "accountId": null,
      "apiKey": null,
      "caller": null,
      "cognitoAuthenticationProvider": null,
      "cognitoAuthenticationType": null,
      "cognitoIdentityPoolId": null,
      "sourceIp": "127.0.0.1",
      "user": null,
      "userAgent": "Custom User Agent String",
      "userArn": null
    },
    "path": "/phagehits",
    "protocol": "HTTP/1.1",
    "requestId": "fc030cd1-4463-405d-8c7c-27111faa2d97",
    "requestTime": "20/Jan/2022:12:33:29 +0000",
    "requestTimeEpoch": 1642682009,
    "resourceId": "123456",
    "resourcePath": "/filter",
    "stage": "Prod"
  },
  "resource": "/filter",
  "stageVariables": null,
  "version": "1.0"
}

models.py

import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, Generic, TypeVar, Optional, Union, List, \
    TypedDict
from uuid import UUID

from dataclass_factory import Schema, NameStyle, validate

# HTTP

BodyT = TypeVar("BodyT")
QueryT = TypeVar("QueryT")
PathT = TypeVar("PathT")


@dataclass
class _BaseHttpResponse(Generic[BodyT]):
    body: BodyT
    status_code: int = 200
    headers: Dict[str, str] = field(default_factory=dict)


@dataclass
class HttpRawResponse(_BaseHttpResponse[str]):
    pass


@dataclass
class HttpResponse(_BaseHttpResponse[BodyT]):
    pass


@dataclass
class HttpRequestContext:
    account_id: str
    api_id: str
    identity: Dict[str, Any]
    path: str
    protocol: str
    request_id: str
    resource_id: str
    resource_path: str
    stage: str


@dataclass
class HttpRequest(Generic[BodyT, QueryT, PathT]):
    headers: Dict[str, str]
    http_method: str
    request_context: HttpRequestContext
    path: str
    resource: str

    body: BodyT
    path_parameters: PathT
    query_string_parameters: QueryT


## REST Base
NumberT = TypeVar("NumberT")


@dataclass
class MinMax(Generic[NumberT]):
    min_value: Optional[NumberT] = None
    max_value: Optional[NumberT] = None


MinMaxFloat = MinMax[float]
MinMaxInt = MinMax[int]
SimpleFilter = Union[None, List, MinMax, Dict[str, MinMaxFloat]]


@dataclass
class Order(Enum):
    asc = "asc"
    desc = "desc"


@dataclass
class PagingRequest:
    offset: int = 0
    limit: int = 20


## REST Concrete
class Stereochemistry(Enum):
    L = "L"
    D = "D"


class Contrast(Enum):
    BLANK = "BLANK"
    POSITIVE = "POSITIVE"
    COMPETITOR = "COMPETITOR"


class SortBy(Enum):
    number_aa = "number_aa"
    counts = "counts"
    id = "id"
    hit_id = "hit_id"
    hit_strength = "hit_strength"
    ic50_nm = "ic50_nm"
    phage_sample_group = "phage_sample_group"
    phage_library = "phage_library"
    cluster = "cluster"
    sequence = "sequence"

    total_aa = "total_aa"
    num_ali_hydro = "num_ali_hydro"
    num_aro_hydro = "num_aro_hydro"
    num_negative_charged = "num_negative_charged"
    num_arg_his = "num_arg_his"
    num_polar = "num_polar"
    active_from = "active_from"


@dataclass
class SortingParams:
    sort_by: SortBy = SortBy.active_from
    order: Order = Order.desc
    sort_key: Optional[str] = None


@dataclass
class SampleFilter(TypedDict, total=False):
    construct: List[UUID]
    stereochemistry: List[Stereochemistry]
    contrast: List[Contrast]
    protein_complex: List[UUID]
    concentration: Optional[MinMaxFloat]
    competitor: str
    screen: List[UUID]
    gene: List[UUID]


class FilterFields(TypedDict, total=False):
    id: Optional[List[UUID]]
    hit_id: Optional[List[str]]
    hit_strength: Optional[MinMaxFloat]
    ic50_nm: Optional[MinMaxFloat]
    phage_sample_group: Optional[List[UUID]]
    cluster: Optional[List[UUID]]
    sequence: Optional[List[str]]
    phage_library: Optional[List[UUID]]
    counts: Optional[Dict[UUID, MinMaxFloat]]
    samples: Optional[SampleFilter]
    sets: Optional[List[UUID]]
    total_aa: Optional[MinMaxInt]
    num_ali_hydro: Optional[MinMaxInt]
    num_aro_hydro: Optional[MinMaxInt]
    num_negative_charged: Optional[MinMaxInt]
    num_arg_his: Optional[MinMaxInt]
    num_polar: Optional[MinMaxInt]


@dataclass
class PhageHitsFilterRequest:
    filters: Optional[FilterFields] = None
    sort: SortingParams = SortingParams()
    pagination: PagingRequest = PagingRequest()


## Schema
class HttpRequestSchema(Schema):
    name_style = NameStyle.camel_lower

    @validate("body", pre=True)
    def json_body(self, data):
        if data is None:
            return None
        return json.loads(data)

    @validate("query_string_parameters", pre=True)
    def fix_empty_query(self, data):
        return data or {}


def create_http_schemas():
    return {
        HttpRequest: HttpRequestSchema(),
        HttpRequestContext: Schema(name_style=NameStyle.camel_lower),
    }


## Method
def handle_filter(
        event: HttpRequest[PhageHitsFilterRequest, Any, Any],
        context: Any,
) -> Any:
    pass