[wip] multipart result + typing fixes

tests/test_formats.py

@@ -4,6 +4,7 @@
 import itertools
 import os
 import uuid
+from typing import TYPE_CHECKING
 from urllib.error import URLError
 from urllib.request import urlopen
 
@@ -18,6 +19,9 @@
 from weaver import formats as f
 from weaver.utils import null, request_extra
 
+if TYPE_CHECKING:
+    from weaver.formats import AnyContentEncoding, FileModeSteamType
+
 _ALLOWED_MEDIA_TYPE_CATEGORIES = [
     "application",
     "archives",
@@ -204,6 +208,7 @@ def test_content_encoding_is_binary(encoding, expect):
     )
 )
 def test_content_encoding_open_parameters(encoding, mode):
+    # type: (AnyContentEncoding, FileModeSteamType) -> None
     result = f.ContentEncoding.open_parameters(encoding, mode)
     if encoding == f.ContentEncoding.UTF_8:
         assert result[0] == mode
@@ -230,7 +235,6 @@ def test_get_format(test_content_type, expected_content_type, expected_content_e
 @pytest.mark.parametrize(
     "test_extension",
     [
-        f.ContentType.APP_OCTET_STREAM,
         f.ContentType.APP_FORM,
         f.ContentType.MULTIPART_FORM,
     ]
@@ -287,7 +291,6 @@ def test_get_format_media_type_from_schema(test_format, expect_media_type):
     itertools.product(
         ["", None],
         [
-            f.ContentType.APP_OCTET_STREAM,
             f.ContentType.APP_FORM,
             f.ContentType.MULTIPART_FORM,
         ]
@@ -299,6 +302,23 @@ def test_get_format_default_no_extension(test_extension, default_content_type):
     assert fmt.extension == ""
 
 
+@pytest.mark.parametrize(
+    ["test_extension", "default_extension"],
+    [
+        ("", f.ContentType.APP_OCTET_STREAM),
+        (None, f.ContentType.APP_OCTET_STREAM),
+        (f.ContentType.APP_OCTET_STREAM, None),
+    ]
+)
+def test_get_format_binary_extension(test_extension, default_extension):
+    """
+    .. versionchanged:: 5.10.0
+    """
+    fmt = f.get_format(test_extension, default=default_extension)
+    assert fmt == Format(f.ContentType.APP_OCTET_STREAM, extension=".bin")
+    assert fmt.extension == ".bin"
+
+
 @pytest.mark.parametrize(
     ["cwl_format", "expect_media_type"],
     [

weaver/datatype.py

@@ -19,7 +19,7 @@
 from io import BytesIO
 from logging import ERROR, INFO, getLevelName, getLogger
 from secrets import compare_digest, token_hex
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 from urllib.parse import urljoin, urlparse
 
 import colander
@@ -210,11 +210,14 @@ class Data(AutoBase):
     """
     def __new__(cls, *args, **kwargs):
         extra_props = set(dir(cls)) - set(dir(DictBase))
-        auto_cls = DictBase.__new__(cls, *args, **kwargs)
+        auto_cls = cast(
+            "AutoBase",
+            DictBase.__new__(cls, *args, **kwargs)
+        )
         for prop in extra_props:
             prop_func = property(
-                lambda self, key: dict.__getitem__(self, key),
-                lambda self, key, value: dict.__setattr__(self, key, value)
+                lambda self, key: dict.__getitem__(self, key),                  # type: ignore
+                lambda self, key, value: dict.__setattr__(self, key, value)     # type: ignore
             )
             default = getattr(auto_cls, prop, None)
             setattr(auto_cls, prop, prop_func)
@@ -908,7 +911,11 @@ def _set_inputs(self, inputs):
         self["inputs"] = inputs
 
     # allows to correctly update list by ref using 'job.inputs.extend()'
-    inputs = property(_get_inputs, _set_inputs, doc="Input values and reference submitted for execution.")
+    inputs = property(
+        _get_inputs,  # type: ignore
+        _set_inputs,  # type: ignore
+        doc="Input values and reference submitted for execution.",
+    )
 
     def _get_outputs(self):
         # type: () -> Optional[ExecutionOutputs]
@@ -918,7 +925,11 @@ def _set_outputs(self, outputs):
         # type: (Optional[ExecutionOutputs]) -> None
         self["outputs"] = outputs
 
-    outputs = property(_get_outputs, _set_outputs, doc="Output transmission modes submitted for execution.")
+    outputs = property(
+        _get_outputs,  # type: ignore
+        _set_outputs,  # type: ignore
+        doc="Output transmission modes submitted for execution.",
+    )
 
     @property
     def user_id(self):
@@ -1119,7 +1130,7 @@ def _get_updated(self):
     created = LocalizedDateTimeProperty(default_now=True)
     started = LocalizedDateTimeProperty()
     finished = LocalizedDateTimeProperty()
-    updated = LocalizedDateTimeProperty(fget=_get_updated)
+    updated = LocalizedDateTimeProperty(fget=_get_updated)  # type: ignore
 
     @property
     def duration(self):
@@ -1179,7 +1190,11 @@ def _set_results(self, results):
         self["results"] = results
 
     # allows to correctly update list by ref using 'job.results.extend()'
-    results = property(_get_results, _set_results, doc="Output values and references that resulted from execution.")
+    results = property(
+        _get_results,  # type: ignore
+        _set_results,  # type: ignore
+        doc="Output values and references that resulted from execution.",
+    )
 
     def _get_exceptions(self):
         # type: () -> List[Union[str, Dict[str, str]]]
@@ -1194,7 +1209,7 @@ def _set_exceptions(self, exceptions):
         self["exceptions"] = exceptions
 
     # allows to correctly update list by ref using 'job.exceptions.extend()'
-    exceptions = property(_get_exceptions, _set_exceptions)
+    exceptions = property(_get_exceptions, _set_exceptions)  # type: ignore
 
     def _get_logs(self):
         # type: () -> List[str]
@@ -1209,7 +1224,7 @@ def _set_logs(self, logs):
         self["logs"] = logs
 
     # allows to correctly update list by ref using 'job.logs.extend()'
-    logs = property(_get_logs, _set_logs)
+    logs = property(_get_logs, _set_logs)  # type: ignore
 
     def _get_tags(self):
         # type: () -> List[Optional[str]]
@@ -1224,7 +1239,7 @@ def _set_tags(self, tags):
         self["tags"] = tags
 
     # allows to correctly update list by ref using 'job.tags.extend()'
-    tags = property(_get_tags, _set_tags)
+    tags = property(_get_tags, _set_tags)  # type: ignore
 
     @property
     def access(self):
@@ -1431,7 +1446,7 @@ def json(self, container=None):  # pylint: disable=W0221,arguments-differ
             "estimatedCompletion": None,
             "percentCompleted": self.progress,
             # new name as per OGC-API, enforced integer
-            # https://github.com/opengeospatial/ogcapi-processes/blob/master/openapi/schemas/processes-core/statusInfo.yaml
+            # https://schemas.opengis.net/ogcapi/processes/part1/1.0/openapi/schemas/statusInfo.yaml
             "progress": int(self.progress),
             "links": self.links(settings, self_link="status")
         }
@@ -1956,9 +1971,11 @@ def _set_id(self, _id):
         # type: (str) -> None
         self["id"] = _id
 
-    id = identifier = property(fget=_get_id, fset=_set_id, doc=(
-        "Unique process identifier with optional version number if it corresponds to an older revision."
-    ))
+    id = identifier = property(
+        _get_id,  # type: ignore
+        _set_id,  # type: ignore
+        doc="Unique process identifier with optional version number if it corresponds to an older revision.",
+    )
 
     @classmethod
     def split_version(cls, process_id):
@@ -2031,7 +2048,11 @@ def _set_desc(self, description):
         # type: (str) -> None
         self["abstract"] = description
 
-    description = abstract = property(fget=_get_desc, fset=_set_desc, doc="Process description.")
+    description = abstract = property(
+        fget=_get_desc,     # type: ignore
+        fset=_set_desc,     # type: ignore
+        doc="Process description.",
+    )
 
     @property
     def keywords(self):

weaver/execute.py

@@ -7,22 +7,67 @@
 from weaver.utils import get_header, parse_kvp
 
 if TYPE_CHECKING:
-    from typing import List, Optional, Tuple
+    from typing import List, Optional, Union, Tuple
 
-    from weaver.typedefs import AnyHeadersContainer, HeadersType
+    from weaver.typedefs import AnyHeadersContainer, HeadersType, Literal
+
+    ExecutionModeAutoType = Literal["auto"]
+    ExecutionModeAsyncType = Literal["async"]
+    ExecutionModeSyncType = Literal["sync"]
+    AnyExecuteMode = Union[
+        ExecutionModeAutoType,
+        ExecutionModeAsyncType,
+        ExecutionModeSyncType,
+    ]
+    ExecuteControlOptionAsyncType = Literal["async-execute"]
+    ExecuteControlOptionSyncType = Literal["sync-execute"]
+    AnyExecuteControlOption = Union[
+        ExecuteControlOptionAsyncType,
+        ExecuteControlOptionSyncType,
+    ]
+    ExecuteReturnPreferenceMinimalType = Literal["minimal"]
+    ExecuteReturnPreferenceRepresentationType = Literal["representation"]
+    AnyExecuteReturnPreference = Union[
+        ExecuteReturnPreferenceMinimalType,
+        ExecuteReturnPreferenceRepresentationType,
+    ]
+    ExecuteResponseDocumentType = Literal["document"]
+    ExecuteResponseRawType = Literal["raw"]
+    AnyExecuteResponse = Union[
+        ExecuteResponseDocumentType,
+        ExecuteResponseRawType,
+    ]
+    ExecuteTransmissionModeReferenceType = Literal["reference"]
+    ExecuteTransmissionModeValueType = Literal["value"]
+    AnyExecuteTransmissionMode = Union[
+        ExecuteTransmissionModeReferenceType,
+        ExecuteTransmissionModeValueType,
+    ]
+    ExecuteCollectionFormatType_STAC = Literal["stac-collection"]
+    ExecuteCollectionFormatType_OGC_COVERAGE = Literal["ogc-coverage-collection"]
+    ExecuteCollectionFormatType_OGC_FEATURES = Literal["ogc-features-collection"]
+    ExecuteCollectionFormatType_OGC_MAP = Literal["ogc-map-collection"]
+    ExecuteCollectionFormatType_GEOJSON = Literal["geojson-feature-collection"]
+    AnyExecuteCollectionFormat = Union[
+        ExecuteCollectionFormatType_STAC,
+        ExecuteCollectionFormatType_OGC_COVERAGE,
+        ExecuteCollectionFormatType_OGC_FEATURES,
+        ExecuteCollectionFormatType_OGC_MAP,
+        ExecuteCollectionFormatType_GEOJSON,
+    ]
 
 LOGGER = logging.getLogger(__name__)
 
 
 class ExecuteMode(Constants):
-    AUTO = "auto"
-    ASYNC = "async"
-    SYNC = "sync"
+    AUTO = "auto"       # type: ExecutionModeAutoType
+    ASYNC = "async"     # type: ExecutionModeAsyncType
+    SYNC = "sync"       # type: ExecutionModeSyncType
 
 
 class ExecuteControlOption(Constants):
-    ASYNC = "async-execute"
-    SYNC = "sync-execute"
+    ASYNC = "async-execute"     # type: ExecuteControlOptionAsyncType
+    SYNC = "sync-execute"       # type: ExecuteControlOptionSyncType
 
     @classmethod
     def values(cls):
@@ -34,58 +79,30 @@ def values(cls):
 
 
 class ExecuteReturnPreference(Constants):
-    MINIMAL = "minimal"
-    REPRESENTATION = "representation"
+    MINIMAL = "minimal"                 # type: ExecuteReturnPreferenceMinimalType
+    REPRESENTATION = "representation"   # type: ExecuteReturnPreferenceRepresentationType
 
 
 class ExecuteResponse(Constants):
-    RAW = "raw"
-    DOCUMENT = "document"
+    RAW = "raw"             # type: ExecuteResponseRawType
+    DOCUMENT = "document"   # type: ExecuteResponseDocumentType
 
 
 class ExecuteTransmissionMode(Constants):
-    VALUE = "value"
-    REFERENCE = "reference"
+    VALUE = "value"             # type: ExecuteTransmissionModeValueType
+    REFERENCE = "reference"     # type: ExecuteTransmissionModeReferenceType
 
 
 class ExecuteCollectionFormat(Constants):
-    STAC = "stac-collection"
-    OGC_COVERAGE = "ogc-coverage-collection"
-    OGC_FEATURES = "ogc-features-collection"
-    OGC_MAP = "ogc-map-collection"
-    GEOJSON = "geojson-feature-collection"
-
-
-if TYPE_CHECKING:
-    from weaver.typedefs import Literal
-
-    AnyExecuteMode = Literal[
-        ExecuteMode.ASYNC,
-        ExecuteMode.SYNC,
-    ]
-    AnyExecuteControlOption = Literal[
-        ExecuteControlOption.ASYNC,
-        ExecuteControlOption.SYNC,
-    ]
-    AnyExecuteResponse = Literal[
-        ExecuteResponse.DOCUMENT,
-        ExecuteResponse.RAW,
-    ]
-    AnyExecuteTransmissionMode = Literal[
-        ExecuteTransmissionMode.REFERENCE,
-        ExecuteTransmissionMode.VALUE,
-    ]
-    AnyExecuteCollectionFormat = Literal[
-        ExecuteCollectionFormat.STAC,
-        ExecuteCollectionFormat.OGC_COVERAGE,
-        ExecuteCollectionFormat.OGC_FEATURES,
-        ExecuteCollectionFormat.OGC_MAP,
-        ExecuteCollectionFormat.GEOJSON,
-    ]
+    STAC = "stac-collection"                    # type: ExecuteCollectionFormatType_STAC
+    OGC_COVERAGE = "ogc-coverage-collection"    # type: ExecuteCollectionFormatType_OGC_COVERAGE
+    OGC_FEATURES = "ogc-features-collection"    # type: ExecuteCollectionFormatType_OGC_FEATURES
+    OGC_MAP = "ogc-map-collection"              # type: ExecuteCollectionFormatType_OGC_MAP
+    GEOJSON = "geojson-feature-collection"      # type: ExecuteCollectionFormatType_GEOJSON
 
 
 def parse_prefer_header_return(headers):
-    # type: (AnyHeadersContainer) -> Optional[ExecuteReturnPreference]
+    # type: (AnyHeadersContainer) -> Optional[AnyExecuteReturnPreference]
     """
     Get the return preference if specified.
     """

weaver/formats.py

@@ -449,7 +449,7 @@ class SchemaRole(Constants):
     ContentType.IMAGE_TIFF: ".tif",  # common alternate to .tiff
     ContentType.ANY: ".*",      # any for glob
     ContentType.APP_DIR: "/",   # force href to finish with explicit '/' to mark directory
-    ContentType.APP_OCTET_STREAM: "",
+    ContentType.APP_OCTET_STREAM: ".bin",
     ContentType.APP_FORM: "",
     ContentType.MULTIPART_FORM: "",
 }