Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/utils/utils.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 | ||||
import base64 | import base64 | ||||
from datetime import datetime | from datetime import datetime | ||||
from typing import List, Optional | from typing import List, Optional | ||||
from swh.storage.interface import PagedResult | from swh.storage.interface import PagedResult | ||||
ENCODING = "utf-8" | ENCODING = "utf-8" | ||||
CURSOR_PREFIX = "TRKJJeGR" | |||||
CURSOR_SUFFIX = "JJEHDNOHT" | |||||
def get_b64_string(source) -> str: | def get_b64_string(source) -> str: | ||||
if type(source) is str: | if type(source) is str: | ||||
source = source.encode(ENCODING) | source = source.encode(ENCODING) | ||||
return base64.b64encode(source).decode("ascii") | return base64.b64encode(source).decode("ascii") | ||||
def get_encoded_cursor(cursor: Optional[str]) -> Optional[str]: | def get_encoded_cursor(cursor: Optional[str]) -> Optional[str]: | ||||
if cursor is None: | if cursor is None: | ||||
return None | return None | ||||
if type(cursor) == str and cursor.isnumeric(): | |||||
# Add two random strings to make cursor a bit more opaque | |||||
# add this only when the cursor is numeric, this check is | |||||
# to avoid making cursor unnecessarily longer in other situations | |||||
# (eg: in snapshot-branches the branch name is used as the cursor) | |||||
cursor = CURSOR_PREFIX + cursor + CURSOR_SUFFIX | |||||
return get_b64_string(cursor) | return get_b64_string(cursor) | ||||
def get_decoded_cursor(cursor: Optional[str]) -> Optional[str]: | def get_decoded_cursor(cursor: Optional[str]) -> Optional[str]: | ||||
if cursor is None: | if cursor is None: | ||||
return None | return None | ||||
return base64.b64decode(cursor, validate=True).decode() | cursor = base64.b64decode(cursor, validate=True).decode() | ||||
if cursor.startswith(CURSOR_PREFIX) and cursor.endswith(CURSOR_SUFFIX): | |||||
# strip the added random strings from left and right | |||||
return cursor[len(CURSOR_PREFIX) : -len(CURSOR_SUFFIX)] | |||||
return cursor | |||||
def get_formatted_date(date: datetime) -> str: | def get_formatted_date(date: datetime) -> str: | ||||
# FIXME, handle error + return other formats | # FIXME, handle error + return other formats | ||||
return date.isoformat() | return date.isoformat() | ||||
def paginated(source: List, first: int, after=0) -> PagedResult: | def paginated(source: List, first: int, after=0) -> PagedResult: | ||||
Show All 15 Lines |