An example of a WIP WebCore resource dispatch implementation for "generic" database-backed collections and resources.

_example.py

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']))

collection.py

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]

resource.py

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)