PK!Q566hydra_client/__init__.py__version__ = "0.1.0" from .client import HydraAdmin PK!*hydra_client/abc.pyfrom __future__ import annotations import abc import typing import requests from . import exceptions from .utils import urljoin if typing.TYPE_CHECKING: from .client import Hydra class AbstractResource(abc.ABC): def __init__(self, resource: AbstractResource = None): if resource is not None: self.session: requests.Session = resource.session self.url: str = resource.url else: self.session = requests.Session() self.url = "" def _request( self, method: str, url: str, params: dict = None, json: dict = None ) -> requests.Response: try: response = self.session.request(method, url, json=json) except ( requests.exceptions.ConnectionError, requests.exceptions.Timeout, ) as exc: raise exceptions.ConnectionError from exc except requests.exceptions.RequestException as exc: raise exceptions.TransportError from exc try: response.raise_for_status() except requests.exceptions.HTTPError as exc: raise exceptions.HTTPError from exc return response class AbstractEndpoint(AbstractResource): def __init__(self, resource: AbstractResource = None): super().__init__(resource) self.url = urljoin(self.url, self.endpoint) @abc.abstractproperty def endpoint(self) -> str: pass PK!2Q66hydra_client/client.pyfrom .abc import AbstractResource from .consent import ConsentRequest from .login import LoginRequest from .version import Version class Hydra(AbstractResource): def __init__(self, url: str): super().__init__() self.url = url class HydraAdmin(Hydra): def login_request(self, challenge: str) -> LoginRequest: return LoginRequest.get(challenge, self) def consent_request(self, challenge: str) -> ConsentRequest: return ConsentRequest.get(challenge, self) def version(self) -> str: return Version.get(self) PK!U( hydra_client/consent.pyfrom __future__ import annotations import typing from .abc import AbstractEndpoint, AbstractResource from .utils import filter_none, urljoin if typing.TYPE_CHECKING: from .client import Hydra class ConsentRequest(AbstractEndpoint): endpoint = "/oauth2/auth/requests/consent" def __init__(self, data: dict, parent: AbstractResource): super().__init__(parent) self.acr = data["acr"] self.challenge = data["challenge"] self.client = data["client"] self.login_challenge = data["login_challenge"] self.login_session_id = data["login_session_id"] self.oidc_context = data["oidc_context"] self.request_url = data["request_url"] self.requested_access_token_audience = data["requested_access_token_audience"] self.requested_scope = data["requested_scope"] self.skip = data["skip"] self.subject = data["subject"] self.url = urljoin(self.url, self.challenge) @classmethod def get(cls, challenge: str, hydra: Hydra) -> ConsentRequest: url = urljoin(hydra.url, cls.endpoint, challenge) response = hydra._request("GET", url) return cls(response.json(), hydra) def accept( self, grant_access_token_audience: typing.Iterable[str] = None, grant_scope: typing.Iterable[str] = None, remember: bool = False, remember_for: int = None, session: dict = None, ) -> str: data = filter_none( { "grant_access_token_audience": grant_access_token_audience, "grant_scope": grant_scope, "remember": remember, "remember_for": remember_for, "session": session, } ) url = urljoin(self.url, "accept") response = self._request("PUT", url, json=data) # TODO: Catch exceptions and raise custom error payload = response.json() return payload["redirect_to"] def reject( self, error: str = None, error_debug: str = None, error_description: str = None, error_hint: str = None, status_code: int = None, ) -> str: url = urljoin(self.url, "reject") data = filter_none( { "error": error, "error_debug": error_debug, "error_description": error_description, "error_hint": error_hint, "status_code": status_code, } ) response = self._request("PUT", url, json=data) # TODO: Catch exceptions and raise custom error payload = response.json() return payload["redirect_to"] PK!+hydra_client/exceptions.pyclass HydraException(Exception): pass class UnboundResourceError(HydraException): pass class HTTPError(HydraException): pass class TransportError(HydraException): pass class ConnectionError(TransportError): pass PK! hydra_client/login.pyfrom __future__ import annotations import typing from .abc import AbstractEndpoint, AbstractResource from .utils import filter_none, urljoin if typing.TYPE_CHECKING: from .client import Hydra class LoginRequest(AbstractEndpoint): endpoint = "/oauth2/auth/requests/login" def __init__(self, data: dict, parent: AbstractResource): super().__init__(parent) self.challenge = data["challenge"] self.client = data["client"] self.oidc_context = data["oidc_context"] self.request_url = data["request_url"] self.requested_access_token_audience = data["requested_access_token_audience"] self.requested_scope = data["requested_scope"] self.session_id = data["session_id"] self.skip = data["skip"] self.subject = data["subject"] self.url = urljoin(self.url, self.challenge) @classmethod def get(cls, challenge: str, hydra: Hydra) -> LoginRequest: url = urljoin(hydra.url, cls.endpoint, challenge) response = hydra._request("GET", url) return cls(response.json(), hydra) def accept( self, acr: str = None, force_subject_identifier: str = None, remember: bool = False, remember_for: int = None, subject: str = None, ) -> str: data = filter_none( { "acr": acr, "force_subject_identifier": force_subject_identifier, "remember": remember, "remember_for": remember_for, "subject": subject, } ) url = urljoin(self.url, "accept") response = self._request("PUT", url, json=data) # TODO: Catch exceptions and raise custom error payload = response.json() return payload["redirect_to"] def reject( self, error: str = None, error_debug: str = None, error_description: str = None, error_hint: str = None, status_code: int = None, ) -> str: url = urljoin(self.url, "reject") data = filter_none( { "error": error, "error_debug": error_debug, "error_description": error_description, "error_hint": error_hint, "status_code": status_code, } ) response = self._request("PUT", url, json=data) # TODO: Catch exceptions and raise custom error payload = response.json() return payload["redirect_to"] PK!%y0hydra_client/utils.pydef filter_none(data: dict) -> dict: return {k: v for k, v in data.items() if v is not None} def urljoin(url: str, *parts: str) -> str: return "/".join( (url.rstrip("/"), *(p.strip("/") for p in parts[:-1]), (parts[-1]).lstrip("/")) ) PK!U hydra_client/version.pyfrom __future__ import annotations import typing import requests from . import exceptions from .abc import AbstractEndpoint from .utils import urljoin if typing.TYPE_CHECKING: from .client import Hydra class Version(AbstractEndpoint): endpoint = "/version" @classmethod def get(cls, hydra: Hydra) -> str: url = urljoin(hydra.url, cls.endpoint) response = hydra._request("GET", url) payload = response.json() return payload["version"] PK!HnHTU"hydra_client-0.2.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!Hd5/%hydra_client-0.2.0.dist-info/METADATAQn0 +|LB $XLv4i#5I%0qa~=gKKtB+FO(C}. E#H6OYIa=F 7X.y PKXhp!J2h%-VAѹ$#+L>ɺƺIlDAzvm8 ^K?&pXFl;{O7x0y Z+Npa 6ا2u|G]R 6' ƿ:Ɨ>ll8X k_C&Y@PK!HhQQ#hydra_client-0.2.0.dist-info/RECORD}ϻ@|\l xaM(i.rh_ 'd_=r`ӺE]>L$;h.cG@(A2?SUd$Mk 'R]YE^>,  8 fP.Ƚ[g©v<$s]#*o׊1Y5u{7t6VfIxTM~.+!Wۻ F.- NӲnՖ9rec2޲q=JXz A1 v܄r^(ʖ*v / ʻw^âHWrt,{~$s yF6hJWL8rwގկh>ogD$]tFTKYnϸYێu:mt}%,UXe-dOB%_^k23}mڗU0^)bjzd  #g՗w0PK!Q566hydra_client/__init__.pyPK!*lhydra_client/abc.pyPK!2Q66=hydra_client/client.pyPK!U( hydra_client/consent.pyPK!+rhydra_client/exceptions.pyPK! hydra_client/login.pyPK!%y0hydra_client/utils.pyPK!U hydra_client/version.pyPK!HnHTU" "hydra_client-0.2.0.dist-info/WHEELPK!Hd5/%"hydra_client-0.2.0.dist-info/METADATAPK!HhQQ#$hydra_client-0.2.0.dist-info/RECORDPK U&