Coverage for ramose/_constants.py: 100%

1# SPDX-FileCopyrightText: 2018-2021 Silvio Peroni <silvio.peroni@unibo.it> 

2# SPDX-FileCopyrightText: 2020-2021 Marilena Daquino <marilena.daquino2@unibo.it> 

3# SPDX-FileCopyrightText: 2022 Davide Brembilla 

4# SPDX-FileCopyrightText: 2024 Ivan Heibi <ivan.heibi2@unibo.it> 

5# SPDX-FileCopyrightText: 2025 Sergei Slinkin 

6# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

7# 

8# SPDX-License-Identifier: ISC 

9 

10from functools import cache 

11 

12from requests import PreparedRequest 

13from requests import Session as _RequestsSession 

14from requests.adapters import HTTPAdapter 

15from requests.auth import AuthBase, HTTPDigestAuth 

16 

17FIELD_TYPE_RE = r"([^\(\s]+)\(([^\)]+)\)" 

18PARAM_NAME = r"{([^{}\(\)]+)}" 

19DEFAULT_HTTP_TIMEOUT = 60 

20HTTP_POOL_SIZE = 100 

21FORMAT_PARTS_WITH_MEDIA_TYPE = 3 

22 

23FORMAT_MEDIA_TYPES = { 

24 "json": "application/json", 

25 "csv": "text/csv", 

26} 

27 

28 

29def media_type_for_format(fmt: str) -> str | None: 

30 return FORMAT_MEDIA_TYPES.get((fmt or "").strip().lower()) 

31 

32 

33_http_session = _RequestsSession() 

34for _scheme in ("http://", "https://"): 

35 _http_session.mount(_scheme, HTTPAdapter(pool_connections=HTTP_POOL_SIZE, pool_maxsize=HTTP_POOL_SIZE)) 

36 

37_backend_auth: dict[str, str] = {} 

38 

39 

40class _FixedAuthorization(AuthBase): 

41 def __init__(self, value: str) -> None: 

42 self.value = value 

43 

44 def __call__(self, r: PreparedRequest) -> PreparedRequest: 

45 r.headers["Authorization"] = self.value 

46 return r 

47 

48 

49@cache 

50def _auth_for(value: str) -> AuthBase: 

51 scheme, _, credentials = value.partition(" ") 

52 if scheme.lower() == "digest": 

53 username, _, password = credentials.partition(":") 

54 return HTTPDigestAuth(username, password) 

55 return _FixedAuthorization(value) 

56 

57 

58def backend_auth(endpoint_url: str) -> AuthBase | None: 

59 """Return the credential configured for a specific SPARQL endpoint, or None.""" 

60 value = _backend_auth.get(endpoint_url) 

61 return _auth_for(value) if value else None