Skip to content

rs_server_common/utils/utils2.md

<< Back to index

This module is used to share common functions between apis endpoints. Split it from utils.py because of dependency conflicts between rs-server-catalog and rs-server-common.

AuthInfo dataclass

User authentication information in KeyCloak.

Source code in docs/rs-server/services/common/rs_server_common/utils/utils2.py 
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class AuthInfo:
    """User authentication information in KeyCloak."""

    # User login (preferred username)
    user_login: str

    # IAM roles
    iam_roles: list[str]

    # Oauth2 attributes and/or custom `config` associated to the API key
    attributes: dict[str, Any]

S3Credentials dataclass

S3 object storage credentials.

Source code in docs/rs-server/services/common/rs_server_common/utils/utils2.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@dataclass
class S3Credentials:
    """S3 object storage credentials."""

    # Access key ID
    access_key_id: str

    # Secret access key
    secret_access_key: str

    # Endpoint URL
    endpoint_url: str

    # Region name
    region_name: str

decorate_sync_async(decorating_context, func)

Decorator for both sync and async functions, see: https://stackoverflow.com/a/68746329

Source code in docs/rs-server/services/common/rs_server_common/utils/utils2.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def decorate_sync_async(decorating_context, func):
    """Decorator for both sync and async functions, see: https://stackoverflow.com/a/68746329"""
    if asyncio.iscoroutinefunction(func):

        async def decorated(*args, **kwargs):
            with decorating_context(*args, **kwargs):
                return await func(*args, **kwargs)

    else:

        def decorated(*args, **kwargs):
            with decorating_context(*args, **kwargs):
                return func(*args, **kwargs)

    return functools.wraps(func)(decorated)

filelock(func, env_var)

Avoid concurrent writing to the database using a file lock.

Parameters:

Name Type Description Default
env_var str

environment variable that defines the folder where to save the lock file.

 required
Source code in docs/rs-server/services/common/rs_server_common/utils/utils2.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def filelock(func, env_var: str):
    """
    Avoid concurrent writing to the database using a file lock.

    Args:
        env_var: environment variable that defines the folder where to save the lock file.
    """

    @functools.wraps(func)
    def with_filelock(*args, **kwargs):
        """Wrap the the call to 'func' inside the lock."""

        # Let's do this only if the RSPY_WORKING_DIR environment variable is defined.
        # Write a .lock file inside this directory.
        try:
            with FileLock(Path(os.environ[env_var]) / f"{env_var}.lock"):
                return func(*args, **kwargs)

        # Else just call the function without a lock
        except KeyError:
            return func(*args, **kwargs)

    return with_filelock

read_response_error(response)

Read and return an HTTP response error detail.

Source code in docs/rs-server/services/common/rs_server_common/utils/utils2.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
def read_response_error(response):
    """Read and return an HTTP response error detail."""

    # Try to read the response detail or error
    try:
        _json = response.json()
        detail = _json.get("detail") or _json.get("description") or _json["error"]

    # If this fail, get the full response content
    except Exception:  # pylint: disable=broad-exception-caught
        detail = response.content.decode("utf-8", errors="ignore")

    return detail

read_streaming_response(response) async

Read a json-formatted streaming response content

Source code in docs/rs-server/services/common/rs_server_common/utils/utils2.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
async def read_streaming_response(response: StreamingResponse) -> Any | None:
    """Read a json-formatted streaming response content"""
    try:
        body = [chunk async for chunk in response.body_iterator]
        splits = map(lambda x: x if isinstance(x, bytes) else x.encode(), body)  # type: ignore[union-attr]
        str_content = b"".join(splits).decode()
        py_content = json.loads(str_content) if str_content else None

        return py_content

    # Reset the StreamingResponse so it can be used again
    finally:
        response.body_iterator = iterate_in_threadpool(iter(body))