-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor(session): dynamic props for get/set
- Loading branch information
1 parent
5c92800
commit c762444
Showing
3 changed files
with
88 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,13 @@ | ||
from django.http import HttpRequest | ||
|
||
from web.core.models import UserFlow | ||
from web.oauth.session import Session as OAuthSession | ||
|
||
|
||
class Session(OAuthSession): | ||
|
||
def __init__(self, request: HttpRequest, reset: bool = False): | ||
self.props["userflow"] = UserFlow | ||
super().__init__(request, reset) | ||
if reset: | ||
self.session["userflow"] = None |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,29 +1,65 @@ | ||
import logging | ||
|
||
from django.db import models | ||
from django.http import HttpRequest | ||
|
||
_OAUTH_CLAIMS = "oauth_claims" | ||
_OAUTH_TOKEN = "oauth_token" | ||
from .models import ClientConfig | ||
|
||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
class Session: | ||
|
||
props = { | ||
"oauth_claims_check": str, | ||
"oauth_claims_eligibility": str, | ||
"oauth_claims_verified": str, | ||
"oauth_config": ClientConfig, | ||
"oauth_redirect_failure": str, | ||
"oauth_redirect_success": str, | ||
"oauth_scopes": str, | ||
"oauth_token": str, | ||
} | ||
|
||
def __init__(self, request: HttpRequest, reset: bool = False): | ||
self.request = request | ||
self.session = request.session | ||
|
||
logger.debug(self.session.keys()) | ||
|
||
def logged_in(request: HttpRequest): | ||
"""Check if the current session has an OAuth token.""" | ||
return bool(oauth_token(request)) | ||
for prop_name in self.props.keys(): | ||
self._make_property(prop_name) | ||
|
||
if reset: | ||
self.oauth_claims_check = "" | ||
self.oauth_claims_eligibility = "" | ||
self.logout() | ||
|
||
def logout(request: HttpRequest): | ||
"""Reset the session claims and tokens.""" | ||
oauth_claims(request, []) | ||
oauth_token(request, "") | ||
def _make_property(self, prop_name): | ||
def _getter(s): | ||
if issubclass(self.props[prop_name], models.Model): | ||
val = s.session.get(prop_name) | ||
cls = self.props[prop_name] | ||
return cls.objects.filter(id=val).first() | ||
else: | ||
return s.session.get(prop_name) | ||
|
||
def _setter(s, value): | ||
if issubclass(self.props[prop_name], models.Model): | ||
s.session[prop_name] = str(getattr(value, "id")) | ||
else: | ||
s.session[prop_name] = value | ||
|
||
def oauth_claims(request: HttpRequest, new_value: list[str] = None) -> str | None: | ||
"""Get the oauth claims from the request's session. Optionally update the value first.""" | ||
if new_value is not None: | ||
request.session[_OAUTH_CLAIMS] = new_value | ||
return request.session.get(_OAUTH_CLAIMS) | ||
prop = property(fget=_getter, fset=_setter) | ||
setattr(self.__class__, prop_name, prop) | ||
|
||
@property | ||
def logged_in(self): | ||
"""Check if the current session has an OAuth token.""" | ||
return bool(self.oauth_token) | ||
|
||
def oauth_token(request: HttpRequest, new_value: str = None) -> str | None: | ||
"""Get the oauth token from the request's session. Optionally update the value first.""" | ||
if new_value is not None: | ||
request.session[_OAUTH_TOKEN] = new_value | ||
return request.session.get(_OAUTH_TOKEN) | ||
def logout(self): | ||
"""Reset the session claims and tokens.""" | ||
self.oauth_claims_verified = "" | ||
self.oauth_token = "" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,20 @@ | ||
from web.core.models.userflow import UserFlow | ||
from web.core.session import Session as CoreSession | ||
|
||
|
||
def userflow() -> UserFlow | None: | ||
return UserFlow.objects.filter(system_name="vital-records").first() | ||
|
||
|
||
class Session(CoreSession): | ||
def __init__(self, request, reset=False): | ||
super().__init__(request, reset) | ||
if reset: | ||
self.userflow = userflow() | ||
if self.userflow: | ||
self.oauth_config = self.userflow.oauth_config | ||
self.oauth_scopes = self.userflow.scopes | ||
self.oauth_claims_check = self.userflow.all_claims | ||
self.oauth_claims_eligibility = self.userflow.eligibility_claim | ||
self.oauth_redirect_failure = self.userflow.redirect_failure | ||
self.oauth_redirect_success = self.userflow.redirect_success |