16367912334

Committed 18 Jul 2025 10:06AM UTC coverage: 76.934% (+0.1%) from 76.806%

Build # 16367912334

Build Type

push

github

Committed by

web-flow

Commit Message

ci: use Ruff as new formatter and linter (#233)

* wip

* pycodestyle

* update dependencies

* skl2onnx

* use ruff

* apply formatter

* apply lint auto fixes

* manually apply lints

* change check

* ruff ci from branch

Run Details

2805 of 3646 relevant lines covered (76.93%)

0.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.78

geoengine/auth.py

"""
Module for encapsulating Geo Engine authentication
"""

from __future__ import annotations

import os
from typing import ClassVar
from uuid import UUID

import geoengine_openapi_client
import urllib3
from dotenv import load_dotenv
from requests.auth import AuthBase

from geoengine.error import GeoEngineException, MethodOnlyAvailableInGeoEnginePro, UninitializedException


class BearerAuth(AuthBase):  # pylint: disable=too-few-public-methods
    """A bearer token authentication for `requests`"""

    __token: str

    def __init__(self, token: str):
        self.__token = token

    def __call__(self, r):
        r.headers["Authorization"] = f"Bearer {self.__token}"
        return r


class Session:
    """
    A Geo Engine session
    """

    __id: UUID
    __user_id: UUID | None = None
    __valid_until: str | None = None
    __server_url: str
    __timeout: int = 60
    __configuration: geoengine_openapi_client.Configuration

    session: ClassVar[Session | None] = None

    def __init__(self, server_url: str, credentials: tuple[str, str] | None = None, token: str | None = None) -> None:
        """
        Initialize communication between this library and a Geo Engine instance

        If credentials or a token are provided, the session will be authenticated.
        Credentials and token must not be provided at the same time.

        optional arguments:
         - `(email, password)` as tuple
         - `token` as a string

        optional environment variables:
         - `GEOENGINE_EMAIL`
         - `GEOENGINE_PASSWORD`
         - `GEOENGINE_TOKEN`
        """

        session = None

        if credentials is not None and token is not None:
            raise GeoEngineException({"message": "Cannot provide both credentials and token"})

        # Auto-generated SessionApi cannot handle dynamically differing return types (SimpleSession or UserSession).
        # Because of that requests must be send manually.
        http = urllib3.PoolManager()
        user_agent = f"geoengine-python/{geoengine_openapi_client.__version__}"

        if credentials is not None:
            session = http.request(
                "POST",
                f"{server_url}/login",
                headers={"User-Agent": user_agent},
                json={"email": credentials[0], "password": credentials[1]},
                timeout=self.__timeout,
            ).json()
        elif "GEOENGINE_EMAIL" in os.environ and "GEOENGINE_PASSWORD" in os.environ:
            session = http.request(
                "POST",
                f"{server_url}/login",
                headers={"User-Agent": user_agent},
                json={"email": os.environ.get("GEOENGINE_EMAIL"), "password": os.environ.get("GEOENGINE_PASSWORD")},
                timeout=self.__timeout,
            ).json()
        elif token is not None:
            session = http.request(
                "GET",
                f"{server_url}/session",
                headers={"User-Agent": user_agent, "Authorization": f"Bearer {token}"},
                timeout=self.__timeout,
            ).json()
        elif "GEOENGINE_TOKEN" in os.environ:
            session = http.request(
                "GET",
                f"{server_url}/session",
                headers={"User-Agent": user_agent, "Authorization": f"Bearer {os.environ.get('GEOENGINE_TOKEN')}"},
                timeout=self.__timeout,
            ).json()
        else:
            session = http.request(
                "POST", f"{server_url}/anonymous", headers={"User-Agent": user_agent}, timeout=self.__timeout
            ).json()

        if "error" in session:
            raise GeoEngineException(session)

        self.__id = session["id"]

        try:
            self.__user_id = session["user"]["id"]
        except KeyError:
            # user id is only present in Pro
            pass
        except TypeError:
            # user is None in non-Pro
            pass

        if "validUntil" in session:
            self.__valid_until = session["validUntil"]

        self.__server_url = server_url
        self.__configuration = geoengine_openapi_client.Configuration(host=server_url, access_token=session["id"])

    def __repr__(self) -> str:
        """Display representation of a session"""
        r = ""
        r += f"Server:              {self.server_url}\n"

        if self.__user_id is not None:
            r += f"User Id:             {self.__user_id}\n"

        r += f"Session Id:          {self.__id}\n"

        if self.__valid_until is not None:
            r += f"Session valid until: {self.__valid_until}\n"

        return r

    @property
    def auth_header(self) -> dict[str, str]:
        """
        Create an authentication header for the current session
        """

        return {"Authorization": "Bearer " + str(self.__id)}

    @property
    def server_url(self) -> str:
        """
        Return the server url of the current session
        """

        return self.__server_url

    @property
    def configuration(self) -> geoengine_openapi_client.Configuration:
        """
        Return the current http configuration
        """

        return self.__configuration

    @property
    def user_id(self) -> UUID:
        """
        Return the user id. Only works in Geo Engine Pro.
        """
        if self.__user_id is None:
            raise MethodOnlyAvailableInGeoEnginePro("User id is only available in Geo Engine Pro")

        return self.__user_id

    def requests_bearer_auth(self) -> BearerAuth:
        """
        Return a Bearer authentication object for the current session
        """

        return BearerAuth(str(self.__id))

    def logout(self):
        """
        Logout the current session
        """

        with geoengine_openapi_client.ApiClient(self.configuration) as api_client:
            session_api = geoengine_openapi_client.SessionApi(api_client)
            session_api.logout_handler(_request_timeout=self.__timeout)


def get_session() -> Session:
    """
    Return the global session if it exists

    Raises an exception otherwise.
    """

    if Session.session is None:
        raise UninitializedException()

    return Session.session


def initialize(server_url: str, credentials: tuple[str, str] | None = None, token: str | None = None) -> None:
    """
    Initialize communication between this library and a Geo Engine instance

    If credentials or a token are provided, the session will be authenticated.
    Credentials and token must not be provided at the same time.

    optional arugments: (email, password) as tuple or token as a string
    optional environment variables: GEOENGINE_EMAIL, GEOENGINE_PASSWORD, GEOENGINE_TOKEN
    optional .env file defining: GEOENGINE_EMAIL, GEOENGINE_PASSWORD, GEOENGINE_TOKEN
    """

    load_dotenv()

    Session.session = Session(server_url, credentials, token)


def reset(logout: bool = True) -> None:
    """
    Resets the current session
    """

    if Session.session is not None and logout:
        Session.session.logout()

    Session.session = None

1	"""
2	Module for encapsulating Geo Engine authentication
3	"""
4
5	from __future__ import annotations	1✔
6
7	import os	1✔
8	from typing import ClassVar	1✔
9	from uuid import UUID	1✔
10
11	import geoengine_openapi_client	1✔
12	import urllib3	1✔
13	from dotenv import load_dotenv	1✔
14	from requests.auth import AuthBase	1✔
15
16	from geoengine.error import GeoEngineException, MethodOnlyAvailableInGeoEnginePro, UninitializedException	1✔
17
18
19	class BearerAuth(AuthBase): # pylint: disable=too-few-public-methods	1✔
20	"""A bearer token authentication for `requests`"""
21
22	__token: str	1✔
23
24	def __init__(self, token: str):	1✔
25	self.__token = token	1✔
26
27	def __call__(self, r):	1✔
28	r.headers["Authorization"] = f"Bearer {self.__token}"	1✔
29	return r	1✔
30
31
32	class Session:	1✔
33	"""
34	A Geo Engine session
35	"""
36
37	__id: UUID	1✔
38	__user_id: UUID \| None = None	1✔
39	__valid_until: str \| None = None	1✔
40	__server_url: str	1✔
41	__timeout: int = 60	1✔
42	__configuration: geoengine_openapi_client.Configuration	1✔
43
44	session: ClassVar[Session \| None] = None	1✔
45
46	def __init__(self, server_url: str, credentials: tuple[str, str] \| None = None, token: str \| None = None) -> None:	1✔
47	"""
48	Initialize communication between this library and a Geo Engine instance
49
50	If credentials or a token are provided, the session will be authenticated.
51	Credentials and token must not be provided at the same time.
52
53	optional arguments:
54	- `(email, password)` as tuple
55	- `token` as a string
56
57	optional environment variables:
58	- `GEOENGINE_EMAIL`
59	- `GEOENGINE_PASSWORD`
60	- `GEOENGINE_TOKEN`
61	"""
62
63	session = None	1✔
64
65	if credentials is not None and token is not None:	1✔
66	raise GeoEngineException({"message": "Cannot provide both credentials and token"})	1✔
67
68	# Auto-generated SessionApi cannot handle dynamically differing return types (SimpleSession or UserSession).
69	# Because of that requests must be send manually.
70	http = urllib3.PoolManager()	1✔
71	user_agent = f"geoengine-python/{geoengine_openapi_client.__version__}"	1✔
72
73	if credentials is not None:	1✔
74	session = http.request(	1✔
75	"POST",
76	f"{server_url}/login",
77	headers={"User-Agent": user_agent},
78	json={"email": credentials[0], "password": credentials[1]},
79	timeout=self.__timeout,
80	).json()
81	elif "GEOENGINE_EMAIL" in os.environ and "GEOENGINE_PASSWORD" in os.environ:	1✔
82	session = http.request(	1✔
83	"POST",
84	f"{server_url}/login",
85	headers={"User-Agent": user_agent},
86	json={"email": os.environ.get("GEOENGINE_EMAIL"), "password": os.environ.get("GEOENGINE_PASSWORD")},
87	timeout=self.__timeout,
88	).json()
89	elif token is not None:	1✔
90	session = http.request(	1✔
91	"GET",
92	f"{server_url}/session",
93	headers={"User-Agent": user_agent, "Authorization": f"Bearer {token}"},
94	timeout=self.__timeout,
95	).json()
96	elif "GEOENGINE_TOKEN" in os.environ:	1✔
97	session = http.request(	1✔
98	"GET",
99	f"{server_url}/session",
100	headers={"User-Agent": user_agent, "Authorization": f"Bearer {os.environ.get('GEOENGINE_TOKEN')}"},
101	timeout=self.__timeout,
102	).json()
103	else:
104	session = http.request(	1✔
105	"POST", f"{server_url}/anonymous", headers={"User-Agent": user_agent}, timeout=self.__timeout
106	).json()
107
108	if "error" in session:	1✔
109	raise GeoEngineException(session)	×
110
111	self.__id = session["id"]	1✔
112
113	try:	1✔
114	self.__user_id = session["user"]["id"]	1✔
115	except KeyError:	1✔
116	# user id is only present in Pro
117	pass	1✔
118	except TypeError:	1✔
119	# user is None in non-Pro
120	pass	1✔
121
122	if "validUntil" in session:	1✔
123	self.__valid_until = session["validUntil"]	1✔
124
125	self.__server_url = server_url	1✔
126	self.__configuration = geoengine_openapi_client.Configuration(host=server_url, access_token=session["id"])	1✔
127
128	def __repr__(self) -> str:	1✔
129	"""Display representation of a session"""
130	r = ""	×
131	r += f"Server: {self.server_url}\n"	×
132
133	if self.__user_id is not None:	×
134	r += f"User Id: {self.__user_id}\n"	×
135
136	r += f"Session Id: {self.__id}\n"	×
137
138	if self.__valid_until is not None:	×
139	r += f"Session valid until: {self.__valid_until}\n"	×
140
141	return r	×
142
143	@property	1✔
144	def auth_header(self) -> dict[str, str]:	1✔
145	"""
146	Create an authentication header for the current session
147	"""
148
149	return {"Authorization": "Bearer " + str(self.__id)}	1✔
150
151	@property	1✔
152	def server_url(self) -> str:	1✔
153	"""
154	Return the server url of the current session
155	"""
156
157	return self.__server_url	1✔
158
159	@property	1✔
160	def configuration(self) -> geoengine_openapi_client.Configuration:	1✔
161	"""
162	Return the current http configuration
163	"""
164
165	return self.__configuration	1✔
166
167	@property	1✔
168	def user_id(self) -> UUID:	1✔
169	"""
170	Return the user id. Only works in Geo Engine Pro.
171	"""
172	if self.__user_id is None:	1✔
173	raise MethodOnlyAvailableInGeoEnginePro("User id is only available in Geo Engine Pro")	×
174
175	return self.__user_id	1✔
176
177	def requests_bearer_auth(self) -> BearerAuth:	1✔
178	"""
179	Return a Bearer authentication object for the current session
180	"""
181
182	return BearerAuth(str(self.__id))	1✔
183
184	def logout(self):	1✔
185	"""
186	Logout the current session
187	"""
188
189	with geoengine_openapi_client.ApiClient(self.configuration) as api_client:	×
190	session_api = geoengine_openapi_client.SessionApi(api_client)	×
191	session_api.logout_handler(_request_timeout=self.__timeout)	×
192
193
194	def get_session() -> Session:	1✔
195	"""
196	Return the global session if it exists
197
198	Raises an exception otherwise.
199	"""
200
201	if Session.session is None:	1✔
202	raise UninitializedException()	1✔
203
204	return Session.session	1✔
205
206
207	def initialize(server_url: str, credentials: tuple[str, str] \| None = None, token: str \| None = None) -> None:	1✔
208	"""
209	Initialize communication between this library and a Geo Engine instance
210
211	If credentials or a token are provided, the session will be authenticated.
212	Credentials and token must not be provided at the same time.
213
214	optional arugments: (email, password) as tuple or token as a string
215	optional environment variables: GEOENGINE_EMAIL, GEOENGINE_PASSWORD, GEOENGINE_TOKEN
216	optional .env file defining: GEOENGINE_EMAIL, GEOENGINE_PASSWORD, GEOENGINE_TOKEN
217	"""
218
219	load_dotenv()	1✔
220
221	Session.session = Session(server_url, credentials, token)	1✔
222
223
224	def reset(logout: bool = True) -> None:	1✔
225	"""
226	Resets the current session
227	"""
228
229	if Session.session is not None and logout:	1✔
230	Session.session.logout()	×
231
232	Session.session = None	1✔

geo-engine / geoengine-python / 16367912334

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous