forms - multipart/form-data & application/application/x-www-form-urlencoded

Author: commonism

aiopenapi3/v20/glue.py

@@ -1,5 +1,6 @@
 from typing import List, Union, cast
 import json
+import urllib.parse
 
 import httpx
 import pydantic
@@ -122,12 +123,15 @@ def _prepare_parameters(self, provided):
             assert isinstance(values, dict)
 
             if spec.in_ == "formData":
-                if "multipart/form-data" not in self.operation.consumes:
-                    raise ValueError(f"operation does not consume form data but parameter {name} is formData")
-                if spec.type == "file":
-                    self.req.files.update(values)
-                else:
+                if "multipart/form-data" in self.operation.consumes:
+                    if spec.type == "file":
+                        self.req.files.update(values)
+                    else:
+                        self.req.data.update(values)
+                elif "application/x-www-form-urlencoded" in self.operation.consumes:
                     self.req.data.update(values)
+                else:
+                    raise ValueError(f"operation does not consume form data but parameter {name} is formData")
 
             if spec.in_ == "path":
                 # The string method `format` is incapable of partial updates,

aiopenapi3/v30/formdata.py

@@ -0,0 +1,178 @@
+import base64
+import quopri
+from typing import List, Tuple, Dict, Union
+from email.mime import multipart, nonmultipart
+from email.message import _unquotevalue, Message
+import collections
+
+
+class MIMEFormdata(nonmultipart.MIMENonMultipart):
+    def __init__(self, keyname, *args, **kwargs):
+        super(MIMEFormdata, self).__init__(*args, **kwargs)
+        self.add_header("Content-Disposition", f'form-data; name="{keyname}"')
+        del self["MIME-Version"]
+
+
+class MIMEMultipart(multipart.MIMEMultipart):
+    def __init__(self):
+        super().__init__("form-data")
+        del self["MIME-Version"]
+
+    def _write_headers(self, generator):
+        pass
+
+
+from .parameter import encode_parameter
+
+
+def parameters_from_multipart(data, media, rbq):
+    params = list()
+    for k in data.__fields_set__:
+        v = getattr(data, k)
+        ct = "text/plain"
+
+        if (p := media.schema_.properties.get(k, None)) is not None:
+            """OpenAPI 3.0 - Special Considerations for multipart Content"""
+            if p.type == "array":
+                p = p.items
+            if p.type == "string" and (
+                p.format in ("binary", "base64") or getattr(p, "contentEncoding", None) is not None
+            ):
+                ct = "application/octet-stream"
+            elif p.type == "object":
+                ct = "application/json"
+
+        if (e := media.encoding.get(k, None)) != None:
+            ct = e.contentType or ct
+            style = e.style or "form"
+            explode = e.explode if e.explode is not None else (True if style == "form" else False)
+            allowReserved = e.allowReserved or False
+            headers = {name: rbq[name] for name in e.headers.keys() if name in rbq}
+        else:
+            allowReserved = False
+            style = "form"
+            explode = True
+            headers = dict()
+
+        m = media.schema_.properties[k]
+        if isinstance(v, list):
+            for i in v:
+                r = encode_parameter(k, i, style, explode, allowReserved, "query", m.items)
+                params.append((k, ct, r, headers, m.items))
+        else:
+            r = encode_parameter(k, v, style, explode, allowReserved, "query", m)
+            params.append((k, ct, r, headers, m))
+    return params
+
+
+def parameters_from_urlencoded(data: "BaseModel", media: "Media"):
+    params = collections.defaultdict(lambda: list())
+    for k in data.__fields_set__:
+        v = getattr(data, k)
+
+        if (e := media.encoding.get(k, None)) != None:
+            explode = e.explode
+            allowReserved = e.allowReserved
+            style = e.style
+        else:
+            explode = True
+            allowReserved = False
+            style = "form"
+
+        m = media.schema_.properties[k]
+        if isinstance(v, list):
+            for i in v:
+                r = encode_parameter(k, i, style, explode, allowReserved, "query", m.items)
+                params[k].append(r)
+        else:
+            r = encode_parameter(k, v, style, explode, allowReserved, "query", m)
+            params[k].append(r)
+    return params
+
+
+def encode_content(data, codec):
+    """
+    … supports all encodings defined in [RFC4648], including “base64” and “base64url”, as well as “quoted-printable” from [RFC2045].
+    :param data:
+    :param codec:
+    :return:
+    """
+    if codec in ["base16", "base32", "base64", "base64url"]:
+        if codec == "base16":
+            r = base64.b16encode(data)
+        elif codec == "base32":
+            r = base64.b32encode(data)
+        elif codec == "base64":
+            r = base64.b64encode(data)
+        elif codec == "base64url":
+            r = base64.urlsafe_b64encode(data).rstrip(b"=")
+        return r.decode()
+    elif codec == "quoted-printable":
+        return quopri.encodestring(data)
+    else:
+        raise ValueError(f"unsupported codec {codec}")
+
+
+def encode_multipart_parameters(fields: List[Tuple[str, str, Union[str, bytes], Dict[str, str], "Schema"]]):
+    """
+    As shown in
+    https://julien.danjou.info/handling-multipart-form-data-python/
+
+    :param fields:
+    :return:
+    """
+    m = MIMEMultipart()
+
+    for (field, ct, value, headers, schema) in fields:
+        type, subtype, params = decode_content_type(ct)
+
+        if type in ["image", "audio", "application"]:
+            if isinstance(value, bytes):
+                v = value
+            else:
+                v = value.encode()
+
+            codec = "base64"
+
+            if hasattr(schema, "contentEncoding"):
+                """OpenAPI 3.1"""
+                if schema.contentEncoding:
+                    codec = schema.contentEncoding
+                    headers["Content-Encoding"] = codec
+            else:
+                """OpenAPI 3.0"""
+
+            data = encode_content(v, codec)
+
+        elif type in ["text", "rfc822"]:
+            data = value
+        else:
+            type, subtype = "text", "plain"
+            data = value
+
+        env = MIMEFormdata(field, type, subtype)
+
+        for header, value in headers.items():
+            env.add_header(header, value)
+
+        for k, v in params:
+            env.set_param(k, v, "Content-Type")
+
+        env.set_payload(data)
+
+        m.attach(env)
+
+    return m
+
+
+def decode_content_type(value: str) -> Tuple[str, str, List[Tuple[str, str]]]:
+    """
+    msg = Message._get_params_preserve({"content-type": value}, header="content-type", failobj=None)
+    ct, *params = list(map(lambda x: (x[0], _unquotevalue(x[1])) if x[0].lower() == x[0] else x, msg))
+    """
+    m = Message()
+    m.add_header("content-type", value)
+    ct, *params = m.get_params()
+
+    type, _, subtype = ct[0].partition("/")
+    return type, subtype, params

aiopenapi3/v30/glue.py

@@ -1,13 +1,16 @@
 from typing import List, Union, cast
 import json
+import urllib.parse
 
 import httpx
 import pydantic
 import pydantic.json
 
+import aiopenapi3.v30.media
 from ..base import SchemaBase, ParameterBase
 from ..request import RequestBase, AsyncRequestBase
 from ..errors import HTTPStatusError, ContentTypeError, ResponseDecodingError, ResponseSchemaError
+from .formdata import parameters_from_multipart, parameters_from_urlencoded, encode_multipart_parameters
 
 
 class Request(RequestBase):
@@ -110,9 +113,18 @@ def _prepare_parameters(self, provided):
           https://spec.openapis.org/oas/v3.0.3#parameter-object
           A unique parameter is defined by a combination of a name and location.
         """
+
         provided = provided or dict()
         possible = {_.name: _ for _ in self.operation.parameters + self.root.paths[self.path].parameters}
 
+        if self.operation.requestBody:
+            rbq = dict()  # requestBody Parameters
+            ct = "multipart/form-data"
+            if ct in self.operation.requestBody.content:
+                for k, v in self.operation.requestBody.content[ct].encoding.items():
+                    rbq.update(v.headers)
+                possible.update(rbq)
+
         parameters = {
             i.name: i.schema_.default for i in filter(lambda x: x.schema_.default is not None, possible.values())
         }
@@ -129,29 +141,30 @@ def _prepare_parameters(self, provided):
             )
 
         path_parameters = {}
-
+        rbqh = dict()
         for name, value in parameters.items():
             spec = possible[name]
             values = spec._encode(name, value)
             assert isinstance(values, dict)
-            if spec.in_ == "path":
+
+            if isinstance(spec, (aiopenapi3.v30.parameter.Header, aiopenapi3.v31.parameter.Header)):
+                rbqh.update(values)
+            elif spec.in_ == "header":
+                self.req.headers.update(values)
+            elif spec.in_ == "path":
                 # The string method `format` is incapable of partial updates,
                 # as such we need to collect all the path parameters before
                 # applying them to the format string.
                 path_parameters.update(values)
-
-            if spec.in_ == "query":
+            elif spec.in_ == "query":
                 self.req.params.update(values)
-
-            if spec.in_ == "header":
-                self.req.headers.update(values)
-
-            if spec.in_ == "cookie":
+            elif spec.in_ == "cookie":
                 self.req.cookies.update(values)
 
         self.req.url = self.req.url.format(**path_parameters)
+        return rbqh
 
-    def _prepare_body(self, data):
+    def _prepare_body(self, data, rbq):
         if not self.operation.requestBody:
             return
 
@@ -173,23 +186,37 @@ def _prepare_body(self, data):
             data = self.api.plugins.message.sending(operationId=self.operation.operationId, sending=data).sending
             self.req.content = data
             self.req.headers["Content-Type"] = "application/json"
-        #        elif "multipart/form-data" in self.operation.requestBody.content:
-        #            """
-        #            https://swagger.io/docs/specification/describing-request-body/multipart-requests/
-        #            """
-        #            pass
-        #        elif "multipart/mixed" in self.operation.requestBody.content:
-        #            pass
-        #        elif "multipart/form-data" in self.operation.requestBody.content:
-        #            pass
-
+        elif (ct := "multipart/form-data") in self.operation.requestBody.content:
+            """
+            https://swagger.io/docs/specification/describing-request-body/multipart-requests/
+            https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#media-type-object
+            """
+            media: aiopenapi3.v30.media.MediaType = self.operation.requestBody.content[ct]
+            if not media.schema_ or not isinstance(data, media.schema_.get_type()):
+                """expect the data to be a model"""
+                raise TypeError((type(data), media.schema_.get_type()))
+
+            params = parameters_from_multipart(data, media, rbq)
+            msg = encode_multipart_parameters(params)
+            self.req.content = msg.as_string()
+            self.req.headers["Content-Type"] = f'{msg.get_content_type()}; boundary="{msg.get_boundary()}"'
+        elif (ct := "application/x-www-form-urlencoded") in self.operation.requestBody.content:
+            self.req.headers["Content-Type"] = ct
+            media: aiopenapi3.v30.media.MediaType = self.operation.requestBody.content[ct]
+            if not media.schema_ or not isinstance(data, media.schema_.get_type()):
+                """expect the data to be a model"""
+                raise TypeError((type(data), media.schema_.get_type()))
+
+            params = parameters_from_urlencoded(data, media)
+            msg = urllib.parse.urlencode(params, doseq=True)
+            self.req.content = msg
         else:
-            raise NotImplementedError()
+            raise NotImplementedError(self.operation.requestBody.content)
 
     def _prepare(self, data, parameters):
         self._prepare_security()
-        self._prepare_parameters(parameters)
-        self._prepare_body(data)
+        rbq = self._prepare_parameters(parameters)
+        self._prepare_body(data, rbq)
 
     def _build_req(self, session):
         req = session.build_request(
@@ -280,7 +307,6 @@ def _process(self, result):
             )
 
         if content_type.lower() == "application/json":
-
             data = ctx.received
             try:
                 data = json.loads(data)

aiopenapi3/v30/parameter.py

@@ -1,4 +1,7 @@
 import enum
+import datetime
+import decimal
+import uuid
 from typing import Union, Optional, Dict, Any
 
 from pydantic import Field, root_validator
@@ -70,7 +73,6 @@ def _encode__matrix(self, name, value, explode):
             else:
                 # ;R=100;G=200;B=150
                 value = f";{value}"
-            pass
         return {name: value}
 
     def _encode__label(self, name, value, explode):
@@ -257,6 +259,18 @@ class Parameter(ParameterBase, _ParameterCodec):
     in_: _In = Field(required=True, alias="in")  # TODO must be one of ["query","header","path","cookie"]
 
 
+def encode_parameter(
+    name: str, value: object, style: str, explode: bool, allowReserved: bool, in_: str, schema_: Schema
+) -> Union[str, bytes]:
+    p = Parameter(name=name, style=style, explode=explode, allowReserved=allowReserved, **{"in": in_, "schema": None})
+    p.schema_ = schema_
+    r = p._encode(name, value)[name]
+    if isinstance(r, (int, float, decimal.Decimal, datetime.datetime, datetime.date, datetime.time, uuid.UUID)):
+        r = str(r)
+    assert isinstance(r, (str, bytes))
+    return r
+
+
 class Header(ParameterBase, _ParameterCodec):
     """