forked from dapr/python-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutil.py
57 lines (40 loc) · 1.52 KB
/
util.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# -*- coding: utf-8 -*-
"""
Copyright (c) Microsoft Corporation.
Licensed under the MIT License.
"""
import re
from datetime import timedelta
# Regex to parse Go Duration datatype, e.g. 4h15m50s
DAPR_DURATION_PARSER = re.compile(r'((?P<hours>\d+)h)?((?P<mins>\d+)m)?((?P<seconds>\d+)s)?')
def convert_from_dapr_duration(duration: str) -> timedelta:
"""Converts Dapr duration format (Go duration format) to datetime.timedelta.
Args:
duration (str): Dapr duration string.
Returns:
:obj:`datetime.delta`: the python datetime object.
"""
matched = DAPR_DURATION_PARSER.match(duration)
if not matched or matched.lastindex == 0:
raise ValueError(f'Invalid Dapr Duartion format: \'{duration}\'')
days = 0.0
hours = 0.0
if matched.group('hours') is not None:
days, hours = divmod(float(matched.group('hours')), 24)
mins = 0.0 if not matched.group('mins') else float(matched.group('mins'))
seconds = 0.0 if not matched.group('seconds') else float(matched.group('seconds'))
return timedelta(
days=days,
hours=hours,
minutes=mins,
seconds=seconds)
def convert_to_dapr_duration(td: timedelta) -> str:
"""Converts date.timedelta to Dapr duration format.
Args:
td (datetime.timedelta): python datetime object.
Returns:
str: dapr duration format string.
"""
total_minutes, secs = divmod(td.total_seconds(), 60.0)
hours, mins = divmod(total_minutes, 60.0)
return f'{hours:.0f}h{mins:.0f}m{secs:.0f}s'