diff --git a/swh/graphql/resolvers/scalars.py b/swh/graphql/resolvers/scalars.py index 61948cf..1769319 100644 --- a/swh/graphql/resolvers/scalars.py +++ b/swh/graphql/resolvers/scalars.py @@ -1,40 +1,41 @@ from ariadne import ScalarType +from swh.graphql.utils import utils from swh.model.fields.hashes import validate_sha1_git # from swh.model.swhids import QualifiedSWHID datetime_scalar = ScalarType("DateTime") swhid_scalar = ScalarType("SWHID") sha1_scalar = ScalarType("Sha1") binary_text_scalar = ScalarType("BinaryText") datetimezone_scalar = ScalarType("DateTimeZone") @datetime_scalar.serializer def serialize_datetime(value): - # FIXME, consider timezone, use core functions - return value.timestamp() + return utils.get_formatted_date(value) @sha1_scalar.serializer def serialize_sha1(value): return value.hex() @sha1_scalar.value_parser def validate_sha1(value): # FIXME, handle the error and raise a Graphql one validate_sha1_git(value) return value @binary_text_scalar.serializer def serialize_binary_text(value): - # FIXME return value.decode("utf-8") @datetimezone_scalar.serializer def serialize_datetimezone(value): - return value.to_datetime().timestamp() + # FIXME, handle error and return None + date = value.to_datetime() + return utils.get_formatted_date(date) diff --git a/swh/graphql/tests/unit/utils/test_utils.py b/swh/graphql/tests/unit/utils/test_utils.py index 88f9a3a..a692a8d 100644 --- a/swh/graphql/tests/unit/utils/test_utils.py +++ b/swh/graphql/tests/unit/utils/test_utils.py @@ -1,53 +1,61 @@ +import datetime + from swh.graphql.utils import utils class TestUtils: def test_b64encode(self): assert utils.b64encode("testing") == "dGVzdGluZw==" def test_get_encoded_cursor_is_none(self): assert utils.get_encoded_cursor(None) is None def test_get_encoded_cursor(self): assert utils.get_encoded_cursor(None) is None assert utils.get_encoded_cursor("testing") == "dGVzdGluZw==" def test_get_decoded_cursor_is_none(self): assert utils.get_decoded_cursor(None) is None def test_get_decoded_cursor(self): assert utils.get_decoded_cursor("dGVzdGluZw==") == "testing" def test_str_to_sha1(self): assert ( utils.str_to_sha1("208f61cc7a5dbc9879ae6e5c2f95891e270f09ef") == b" \x8fa\xccz]\xbc\x98y\xaen\\/\x95\x89\x1e'\x0f\t\xef" ) + def test_get_formatted_date(self): + date = datetime.datetime( + 2015, 8, 4, 22, 26, 14, 804009, tzinfo=datetime.timezone.utc + ) + assert utils.get_formatted_date(date) == "2015-08-04T22:26:14.804009+00:00" + def test_paginated(self): source = [1, 2, 3, 4, 5] response = utils.paginated(source, first=50) assert response.results == source assert response.next_page_token is None def test_paginated_first_arg(self): source = [1, 2, 3, 4, 5] response = utils.paginated(source, first=2) assert response.results == source[:2] assert response.next_page_token == "2" def test_paginated_after_arg(self): source = [1, 2, 3, 4, 5] response = utils.paginated(source, first=2, after="2") assert response.results == [3, 4] assert response.next_page_token == "4" response = utils.paginated(source, first=2, after="3") assert response.results == [4, 5] assert response.next_page_token is None def test_paginated_endcursor_outside(self): source = [1, 2, 3, 4, 5] response = utils.paginated(source, first=2, after="10") assert response.results == [] assert response.next_page_token is None diff --git a/swh/graphql/utils/utils.py b/swh/graphql/utils/utils.py index c489ab7..77748a7 100644 --- a/swh/graphql/utils/utils.py +++ b/swh/graphql/utils/utils.py @@ -1,43 +1,49 @@ import base64 +from datetime import datetime from typing import List from swh.storage.interface import PagedResult def b64encode(text: str) -> str: return base64.b64encode(bytes(text, "utf-8")).decode("utf-8") def get_encoded_cursor(cursor: str) -> str: if cursor is None: return None return b64encode(cursor) def get_decoded_cursor(cursor: str) -> str: if cursor is None: return None return base64.b64decode(cursor).decode("utf-8") def str_to_sha1(sha1: str) -> bytearray: # FIXME, use core function return bytearray.fromhex(sha1) +def get_formatted_date(date: datetime) -> str: + # FIXME, handle error + return other formats + return date.isoformat() + + def paginated(source: List, first: int, after=0) -> PagedResult: """ Pagination at the GraphQL level This is a temporary fix and inefficient. Should eventually be moved to the backend (storage) level """ # FIXME, handle data errors here after = 0 if after is None else int(after) end_cursor = after + first results = source[after:end_cursor] next_page_token = None if len(source) > end_cursor: next_page_token = str(end_cursor) return PagedResult(results=results, next_page_token=next_page_token)