Skip to content

Instantly share code, notes, and snippets.

@amcgregor
Last active July 10, 2023 13:04
Show Gist options
  • Save amcgregor/a0ac39d9cafca3c76277ccff2aa7bed3 to your computer and use it in GitHub Desktop.
Save amcgregor/a0ac39d9cafca3c76277ccff2aa7bed3 to your computer and use it in GitHub Desktop.
An example of a WIP WebCore resource dispatch implementation for "generic" database-backed collections and resources.
from webob.exc import HTTPConflict
from web.dispatch.resource.action import Action
from web.dispatch.resource.generic import GenericCollection as Collection
from web.secuity import when
from web.gilt.helper import _, L_
if __debug__: from web.app.static import Static
def mapped_icon(name): # TODO: Split this out.
"""Provide a property dynamically mapping a 'short identifier' to a real icon name."""
@property
def icon(self):
return mapped_icon.mapping[name]
return icon
mapped_icon.mapping = { # Material UI Icons
'accounts': 'contacts',
'account': 'account_circle',
'account.masquerade': 'supervisor_account',
'sessions': 'date_range',
'session': 'event',
}
class Root:
"""The website or application root; the base of the endpoint hierarchy."""
if __debug__: static = Static('public') # Serve static on-disk files from ./public but only in local development.
@when(when.administrator)
class account(Collection):
"""A collection of user accounts.
POST to this collection to create an account. Users may DELETE their own account, with a shortcut representing
the active user at: /account/me ('me' is not an otherwise valid identifier)
Only administrators have access to large portions of this branch of endpoints, however the active user has
limited access to their own record, specifically.
"""
__label__ = L_("Accounts")
__icon__ = mapped_icon('accounts')
__model__ = Account
@property
def me(self):
"""The active, authenticated user account. This pseudo-record will not appear in listings directly."""
return self.__resource__(self._ctx, self, self._ctx.account)
@when(when.administrator | when.me, inherit=False)
class __resource__(Collection.__resource__):
"""A representation of a user account."""
@property
def __icon__(self):
if self._record.id == (self._ctx.session.formerly or self._ctx.session.identity):
return mapped_icon.mapping['account.own']
return mapped_icon.mapping['account']
def get(self):
"""Depending on Accept, return the detail view of an account or an 'account settings' modal fragment."""
raise NotImplementedError()
def post(self, **data):
"""Update an account from the 'account settings' modal, committed on submission."""
raise NotImplementedError()
@when(when.administrator) # Only admnistrators have access to the "detail view" of accounts.
def patch(self, **data):
"""Update an account from the detail view form committed live."""
raise NotImplementedError()
def delete(self):
"""Delete this user account."""
raise NotImplementedError()
@when(when.anonymous | when.administrative)
class verify(Action):
def get(self, token:Optional[str]=None, /):
raise NotImplementedError()
def post(self, token:Optional[str]=None):
"""Mark an account as e-mail address verified.
A token must be supplied unless administrative.
"""
if token: ... # TODO: Validate the token.
elif not when(when.administrative)(self._ctx):
raise HTTPNotAuthorized()
# Update account details.
self._ctx.account.update_one(verified__now=True)
@when(when.administrator, inherit=False)
class masquerade(Action):
"""Masquerade (temporarily become) the target user, as if you had authenticated as them."""
__label__ = L_("Masquerade")
__icon__ = mapped_icon('account.masquerade')
__rel__ = 'confirm'
def post(self):
self._ctx.session.formerly = self._ctx.session.identity # Record who we were so we can undo this action.
self._ctx.session.identity = self._collection._record._id # The collection of an action is a resource.
return {'ok': True, 'message': _("Now masquerading as \"{}\".").format(self._collection._record)}
def delete(self):
if self._ctx.session.identity != self._collection._record:
raise HTTPConflict(_("Not masquerading as this user."))
self._ctx.session.identity = self._ctx.session.formerly
del self._ctx.session.formerly
return {'ok': True, 'message': _("Masquerade lifted.")}
class session(Collection):
"""A collection of active sessions.
POST to this collection to create a new session and authenticate that session. A session owned by the active
user can be invalidated by way of DELETE, with the active session accessible as: /session/current
"""
__label__ = L_("Sessions")
__icon__ = mapped_icon('sessions')
__model__ = Session # The default key is the id of the record.
@property
def current(self):
"""An easy accessor for the current, active session. This will not appear in collection listings."""
return self.__resource__(self._ctx, self, self.__model__[self._ctx.session.id])
@when(when.administrator | when.own)
class __resource__(Collection.__resource__):
"""A representation of an interactive session."""
@property
def __icon__(self):
if self._record.id == self._ctx.session.id:
return mapped_icon.mapping['session.own']
return mapped_icon.mapping['session']
@when(when.anonymous)
def post(self, identity:str, credential:Optional[str]=None):
"""Perform authentication against the provided identity and optional credential, populating a session.
This will perform lookup and request an appropriate additional credential or redirect to perform foreign
SSO, WebAuthN, or OTP generation and request.
"""
raise NotImplementedError()
@when(when.administrator | when.own)
def delete(self):
"""Invalidate all sessions or all sessions for the active user if not administrative."""
if when.administrator(self._ctx):
result = self.__model__.delete_many()
return dict(ok=True, message=_("Invalidated all sessions.") if result['deletedCount'] else _("No sessions to invalidate."))
result = self.__model__.delete_many(identity=self._ctx.session.identity)
return dict(ok=True, message=__("Invalidated a session.", "Invalidated {} sessions.").format(result['deletedCount']))
class GenericCollection(Collection):
"""A generic declarative collection.
Utilizes the __model__ attribute as a mapping to resolve record lookups as a class mapping, e.g.:
self.__model__[identifier]
The result of this lookup (if successful) will be passed to the __resource__ constructor.
"""
__resource__: GenericResource = GenericResource
__model__: Mapping # The model class to utilize when constructing or loading records.
__key__: str # The model instance attribute to use when updating records.
__columns__ = group(Column, "Enumerate all column attributes of this collection that should be visible.")
__indexes__ = group(Index, "Enumerate all column attributes of this collection that should be visible.")
__filters__ = group(Filter, "Enumerate all column attributes of this collection that should be visible.")
def __query__(self, **kw):
"""Return an iterable query object for the chosen model."""
...
def __getitem__(self, identifier):
"""Load the identified resource from backing data storage."""
return self.__model__[identifier]
from inspect import getmembers
from operator import methodcaller
from typing import Any, Callable, Iterable
from web.dispatch.resource import Resource
class GenericResource(Resource):
@property
def __actions__(self) -> Iterable[Action]:
"""Enumerate all actions associated with the record under consideration."""
def condition(action:Action):
if not issubclass(a, Action): return False
instance = action(self._ctx, self, self._record) # Instantiate for validation.
return bool(instance) # Allow the action's __bool__ method to determine availability.
yield from getmembers(self, condition)
__persist__: Callable[(object, ), Any] = methodcaller('save') # Default persistence method to call is 'save'.
__template__: str # The "view" to utilize when requesting the resource as HTML.
def get(self):
"""Retrieve the record rendered by template view."""
# TODO: Content negotiation; possibly move this to the record classes themselves and just return self._record?
return self.__template__, self._record
def post(self, **data):
"""Update using simple direct attribute assignments."""
protected = {k for k in data if k.startswith('_')}
if protected: raise AttributeError(f"Unwilling to assign a value to a protected attribute or attributes: {protected}")
for k, v in data.items(): setattr(self._record, k, v)
self.__persist__(self._record)
def put(self, **data):
"""Replace whole record through direct assignment, preserving ID."""
protected = {k for k in data if k.startswith('_')}
if protected: raise AttributeError(f"Unwilling to assign a value to a protected attribute or attributes: {protected}")
self._record = self._record.__class__(id=self._record.id, **data)
self.__persist__(self._record)
def patch(self, **data):
... # Update record parametrically.
self.__persist__(self._record)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment