Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/resolvers/scalars.py
| # Copyright (C) 2022 The Software Heritage developers | # Copyright (C) 2022 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| from datetime import datetime | from datetime import datetime | ||||
| from ariadne import ScalarType | from ariadne import ScalarType | ||||
| from swh.graphql.utils import utils | from swh.graphql.utils import utils | ||||
| from swh.model import hashutil | |||||
| from swh.model.model import TimestampWithTimezone | from swh.model.model import TimestampWithTimezone | ||||
| from swh.model.swhids import CoreSWHID | from swh.model.swhids import CoreSWHID | ||||
| datetime_scalar = ScalarType("DateTime") | datetime_scalar = ScalarType("DateTime") | ||||
| swhid_scalar = ScalarType("SWHID") | swhid_scalar = ScalarType("SWHID") | ||||
| id_scalar = ScalarType("ID") | id_scalar = ScalarType("ID") | ||||
| content_hash_scalar = ScalarType("ContentHash") | |||||
| @id_scalar.serializer | @id_scalar.serializer | ||||
| def serialize_id(value): | def serialize_id(value): | ||||
| if type(value) is bytes: | if type(value) is bytes: | ||||
| return value.hex() | return value.hex() | ||||
| return value | return value | ||||
| Show All 11 Lines | |||||
| @swhid_scalar.value_parser | @swhid_scalar.value_parser | ||||
| def validate_swhid(value): | def validate_swhid(value): | ||||
| return CoreSWHID.from_string(value) | return CoreSWHID.from_string(value) | ||||
| @swhid_scalar.serializer | @swhid_scalar.serializer | ||||
| def serialize_swhid(value): | def serialize_swhid(value): | ||||
| return str(value) | return str(value) | ||||
| @content_hash_scalar.value_parser | |||||
| def validate_content_hash(value): | |||||
| try: | |||||
| hash_type, hash_string = value.split(":") | |||||
| hash_value = hashutil.hash_to_bytes(hash_string) | |||||
| except ValueError as e: | |||||
| # FIXME, log this error | |||||
| raise AttributeError("Invalid content checksum", e) | |||||
| except Exception as e: | |||||
| # FIXME, log this error | |||||
| raise AttributeError("Invalid content checksum", e) | |||||
| # FIXME, add validation for the hash_type | |||||
| return hash_type, hash_value | |||||