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_SUFIX = "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_SUFIX | |||||
| 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() | # strip the added random strings from left and right | ||||
| return ( | |||||
| base64.b64decode(cursor, validate=True) | |||||
| .decode() | |||||
| .lstrip(CURSOR_PREFIX) | |||||
| .rstrip(CURSOR_SUFIX) | |||||
| ) | |||||
| 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 | |||||