support duration format

open_alchemy/facades/sqlalchemy/simple.py

@@ -152,7 +152,7 @@ def _handle_number(*, artifacts: oa_types.SimplePropertyArtifacts) -> types.Numb
 
 def _handle_string(
     *, artifacts: oa_types.SimplePropertyArtifacts
-) -> typing.Union[types.String, types.Binary, types.Date, types.DateTime]:
+) -> typing.Union[types.String, types.Binary, types.Date, types.DateTime, types.Interval]:
     """
     Handle artifacts for an string type.
 
@@ -173,6 +173,8 @@ def _handle_string(
         return types.Date()
     if artifacts.open_api.format == "date-time":
         return types.DateTime()
+    if artifacts.open_api.format == "duration":
+        return types.Interval()
     if artifacts.open_api.max_length is None:
         return types.String()
     return types.String(length=artifacts.open_api.max_length)

open_alchemy/facades/sqlalchemy/types.py

@@ -16,6 +16,7 @@
 Binary = sqlalchemy.LargeBinary
 Date = sqlalchemy.Date
 DateTime = sqlalchemy.DateTime
+Interval = sqlalchemy.Interval
 Boolean = sqlalchemy.Boolean
 JSON = sqlalchemy.JSON
 Relationship = orm.RelationshipProperty

open_alchemy/helpers/custom_python_types.py

@@ -0,0 +1,48 @@
+import datetime
+import re
+
+class duration(datetime.timedelta):
+    # duration (ISO 8601)
+    def fromisoformat(duration_string):
+        # adopted from: https://rgxdb.com/r/SA5E91Y
+        regex_str = r'^P(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?)?$'
+        validation = re.match(regex_str, duration_string)
+        if validation:
+            groups = validation.groups()
+
+            # timedelta does not support years and months
+            # => approximate, since the actual calendar time span is not known
+            years_in_days = int(groups[0]) * 365 if groups[0] else 0
+            months_in_days = int(groups[1]) * 30 if groups[1] else 0
+
+            timedelta = {
+                'days': years_in_days + months_in_days + (int(groups[2]) if groups[2] else 0),
+                'hours': int(groups[3]) if groups[3] else 0,
+                'minutes': int(groups[4]) if groups[4] else 0,
+                'seconds': int(groups[5]) if groups[5] else 0
+            }
+
+            return datetime.timedelta(**timedelta)
+        else:
+            raise ValueError(f'Invalid isoformat string: {duration_string!r}')
+
+    def isoformat(td_object: datetime.timedelta) -> str:
+        def zero_is_empty(int_to_str, concat):
+            if int_to_str != 0:
+                return str(int_to_str) + concat
+            else:
+                return ''
+
+        PY = td_object.days // 365
+        PM = (td_object.days - PY * 365) // 30
+        PD = (td_object.days - PY * 365 - PM * 30)
+
+        P = [zero_is_empty(PY,'Y'), zero_is_empty(PM,'M'), zero_is_empty(PD,'D')]
+
+        TS = td_object.seconds
+        TH, TS = divmod(TS, 3600)
+        TM, TS = divmod(TS, 60)
+
+        T = [zero_is_empty(TH,'H'), zero_is_empty(TM,'M'), zero_is_empty(TS,'S')]
+
+        return 'P' + ''.join(P) + 'T' + ''.join(T)    

open_alchemy/helpers/oa_to_py_type.py

@@ -5,7 +5,7 @@
 
 from open_alchemy import exceptions
 from open_alchemy import types
-
+from open_alchemy.helpers import custom_python_types
 
 def convert(
     *, value: types.TColumnDefault, type_: str, format_: typing.Optional[str]
@@ -37,6 +37,11 @@ def convert(
             return datetime.datetime.fromisoformat(value)
         except ValueError as exc:
             raise exceptions.MalformedSchemaError("Invalid date-time string.") from exc
+    if isinstance(value, str) and format_ == "duration":
+        try:
+            return custom_python_types.duration.fromisoformat(value)
+        except ValueError as exc:
+            raise exceptions.MalformedSchemaError("Invalid duration string.") from exc
     if isinstance(value, str) and format_ == "binary":
         return value.encode()
     if format_ == "double":

open_alchemy/models_file/artifacts/type_.py

@@ -9,6 +9,7 @@
     "binary": "bytes",
     "date": "datetime.date",
     "date-time": "datetime.datetime",
+    "duration": "custom_python_types.duration"
 }
 
 
@@ -141,6 +142,8 @@ def typed_dict(*, artifacts: schemas_artifacts.types.TAnyPropertyArtifacts) -> s
             model_type = model_type.replace("datetime.date", "str")
         if artifacts.open_api.format == "date-time":
             model_type = model_type.replace("datetime.datetime", "str")
+        if artifacts.open_api.format == "duration":
+            model_type = model_type.replace("custom_python_types.duration", "str")
 
     return model_type
 

open_alchemy/models_file/models/__init__.py

@@ -31,6 +31,8 @@ def generate(*, models: typing.List[str]) -> str:
             break
         if "datetime." in model:
             imports.add("datetime")
+        if "custom_python_types." in model:
+             imports.add("open_alchemy.helpers.custom_python_types as custom_python_types")
 
     template = jinja2.Template(_TEMPLATE, trim_blocks=True)
     return template.render(

open_alchemy/types.py

@@ -3,8 +3,9 @@
 import dataclasses
 import datetime
 import enum
-import typing
+from open_alchemy.helpers import custom_python_types
 
+import typing
 try:  # pragma: no cover
     from typing import Literal  # pylint: disable=unused-import
     from typing import Protocol
@@ -14,6 +15,7 @@
     from typing_extensions import Protocol  # type: ignore
     from typing_extensions import TypedDict  # type: ignore
 
+
 Schema = typing.Dict[str, typing.Any]
 Schemas = typing.Dict[str, Schema]
 TKwargs = typing.Dict[str, typing.Any]
@@ -119,7 +121,7 @@ class Index(_IndexBase, total=False):
 AnyIndex = typing.Union[ColumnList, ColumnListList, Index, IndexList]
 TColumnDefault = typing.Optional[typing.Union[str, int, float, bool]]
 TPyColumnDefault = typing.Optional[
-    typing.Union[str, int, float, bool, bytes, datetime.date, datetime.datetime]
+    typing.Union[str, int, float, bool, bytes, datetime.date, datetime.datetime, custom_python_types.duration]
 ]
 
 

open_alchemy/utility_base/from_dict/simple.py

@@ -4,7 +4,7 @@
 
 from ... import exceptions
 from ... import types as oa_types
-from ...helpers import peek
+from ...helpers import peek, custom_python_types
 from .. import types
 
 
@@ -76,6 +76,8 @@ def _handle_string(
         return datetime.date.fromisoformat(value)
     if format_ == "date-time":
         return datetime.datetime.fromisoformat(value)
+    if format_ == "duration":
+        return custom_python_types.duration.fromisoformat(value)
     if format_ == "binary":
         return value.encode()
     return value

open_alchemy/utility_base/to_dict/simple.py

@@ -4,7 +4,7 @@
 
 from ... import exceptions
 from ... import types as oa_types
-from ...helpers import peek
+from ...helpers import peek, custom_python_types
 from .. import types
 
 
@@ -79,6 +79,13 @@ def _handle_string(value: types.TSimpleCol, *, schema: oa_types.Schema) -> str:
                 "values."
             )
         return value.isoformat()
+    if format_ == "duration":
+        if not isinstance(value, custom_python_types.duration):
+            raise exceptions.InvalidInstanceError(
+                "String type columns with duration format must have duration "
+                "values."
+            )
+        return value.isoformat()
     if format_ == "binary":
         if not isinstance(value, bytes):
             raise exceptions.InvalidInstanceError(

open_alchemy/utility_base/types.py

@@ -4,6 +4,7 @@
 import typing
 
 from .. import types as oa_types
+from open_alchemy.helpers import custom_python_types
 
 # Types for converting to dictionary
 TSimpleDict = typing.Union[int, float, str, bool]
@@ -15,7 +16,7 @@
 TComplexDict = typing.Union[TOptObjectDict, TOptArrayDict]
 TAnyDict = typing.Union[TComplexDict, TOptSimpleDict]
 # Types for converting from a dictionary
-TStringCol = typing.Union[str, bytes, datetime.date, datetime.datetime]
+TStringCol = typing.Union[str, bytes, datetime.date, datetime.datetime, custom_python_types.duration]
 TSimpleCol = typing.Union[int, float, TStringCol, bool]
 TOptSimpleCol = typing.Optional[TSimpleCol]
 TObjectCol = typing.Any  # pylint: disable=invalid-name