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 
32
33
34
35
36
37
38
39
40
41
42
43
@dataclass
class AuthInfo:
    """User authentication information in KeyCloak."""

    # User login (preferred username)
    user_login: str

    # IAM roles
    iam_roles: list[str]

    # Configuration associated to the API key (not implemented for now)
    apikey_config: dict

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 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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

log_http_exception(logger, status_code, detail=None, headers=None, exception_type=HTTPException)

Log error and return an HTTP exception to be raised by the caller

Source code in docs/rs-server/services/common/rs_server_common/utils/utils2.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def log_http_exception(
    logger,
    status_code: Annotated[
        int,
        Doc(
            """
                HTTP status code to send to the client.
                """,
        ),
    ],
    detail: Annotated[
        Any,
        Doc(
            """
                Any data to be sent to the client in the `detail` key of the JSON
                response.
                """,
        ),
    ] = None,
    headers: Annotated[
        dict[str, str] | None,
        Doc(
            """
                Any headers to send to the client in the response.
                """,
        ),
    ] = None,
    exception_type: type[HTTPException] = HTTPException,
) -> type[HTTPException]:
    """Log error and return an HTTP exception to be raised by the caller"""
    logger.error(detail)
    return exception_type(status_code, detail, headers)  # type: ignore

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 
80
81
82
83
84
85
86
87
88
89
90
91
92
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