refactor process handler, and make a PipeState pydantic basemodel object

src/pyff/api.py

@@ -1,6 +1,7 @@
 import importlib
 import threading
 from datetime import datetime, timedelta
+from enum import Enum
 from json import dumps
 from typing import Any, Dict, Generator, Iterable, List, Mapping, Optional, Tuple
 
@@ -21,7 +22,7 @@
 from pyff.constants import config
 from pyff.exceptions import ResourceException
 from pyff.logs import get_log
-from pyff.pipes import plumbing
+from pyff.pipes import PipeState, plumbing
 from pyff.repo import MDRepository
 from pyff.resource import Resource
 from pyff.samlmd import entity_display_name
@@ -153,16 +154,36 @@ def request_handler(request: Request) -> Response:
     return r
 
 
-def process_handler(request: Request) -> Response:
+class ContentNegPolicy(Enum):
+    extension = 'extension'  # current default
+    adaptive = 'adaptive'
+    header = 'header'  # future default
+
+
+def _process_content_negotiate(
+    policy: ContentNegPolicy, alias: str, path: Optional[str], pfx, request: Request
+) -> Tuple[MediaAccept, Optional[str], Optional[str]]:
     """
-    The main request handler for pyFF. Implements API call hooks and content negotiation.
+    Determine requested content type, based on policy, Accept request header and path extension.
 
-    :param request: the HTTP request object
-    :return: the data to send to the client
+    content_negotiation_policy is one of three values:
+
+        1. extension - current default, inspect the path and if it ends in
+           an extension, e.g. .xml or .json, always strip off the extension to
+           get the entityID and if no accept header or a wildcard header, then
+           use the extension to determine the return Content-Type.
+
+        2. adaptive - only if no accept header or if a wildcard, then inspect
+           the path and if it ends in an extension strip off the extension to
+           get the entityID and use the extension to determine the return
+           Content-Type.
+
+        3. header - future default, do not inspect the path for an extension and
+           use only the Accept header to determine the return Content-Type.
     """
     _ctypes = {'xml': 'application/samlmetadata+xml;application/xml;text/xml', 'json': 'application/json'}
 
-    def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional[str]]:
+    def _split_path(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional[str]]:
         """ Split a path into a base component and an extension. """
         if x is not None:
             x = x.strip()
@@ -178,6 +199,45 @@ def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional
 
         return x, None
 
+    # TODO - sometimes the client sends > 1 accept header value with ','.
+    accept = str(request.accept).split(',')[0]
+    valid_accept = accept and not ('application/*' in accept or 'text/*' in accept or '*/*' in accept)
+
+    path_no_extension, extension = _split_path(path, True)
+    accept_from_extension = accept
+    if extension:
+        accept_from_extension = _ctypes.get(extension, accept)
+
+    if policy == ContentNegPolicy.extension:
+        path = path_no_extension
+        if not valid_accept:
+            accept = accept_from_extension
+    elif policy == ContentNegPolicy.adaptive:
+        if not valid_accept:
+            path = path_no_extension
+            accept = accept_from_extension
+
+    if not accept:
+        log.warning('Could not determine accepted response type')
+        raise exc.exception_response(400)
+
+    q: Optional[str]
+    if pfx and path:
+        q = f'{{{pfx}}}{path}'
+        path = f'/{alias}/{path}'
+    else:
+        q = path
+
+    return MediaAccept(accept), path, q
+
+
+def process_handler(request: Request) -> Response:
+    """
+    The main request handler for pyFF. Implements API call hooks and content negotiation.
+
+    :param request: the HTTP request object
+    :return: the data to send to the client
+    """
     log.debug(f'Processing request: {request}')
 
     if request.matchdict is None:
@@ -215,83 +275,46 @@ def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional
         if pfx is None:
             raise exc.exception_response(404)
 
-    # content_negotiation_policy is one of three values:
-    # 1. extension - current default, inspect the path and if it ends in
-    #    an extension, e.g. .xml or .json, always strip off the extension to
-    #    get the entityID and if no accept header or a wildcard header, then
-    #    use the extension to determine the return Content-Type.
-    #
-    # 2. adaptive - only if no accept header or if a wildcard, then inspect
-    #    the path and if it ends in an extension strip off the extension to
-    #    get the entityID and use the extension to determine the return
-    #    Content-Type.
-    #
-    # 3. header - future default, do not inspect the path for an extension and
-    #    use only the Accept header to determine the return Content-Type.
-    policy = config.content_negotiation_policy
-
-    # TODO - sometimes the client sends > 1 accept header value with ','.
-    accept = str(request.accept).split(',')[0]
-    valid_accept = accept and not ('application/*' in accept or 'text/*' in accept or '*/*' in accept)
-
-    new_path: Optional[str] = path
-    path_no_extension, extension = _d(new_path, True)
-    accept_from_extension = accept
-    if extension:
-        accept_from_extension = _ctypes.get(extension, accept)
-
-    if policy == 'extension':
-        new_path = path_no_extension
-        if not valid_accept:
-            accept = accept_from_extension
-    elif policy == 'adaptive':
-        if not valid_accept:
-            new_path = path_no_extension
-            accept = accept_from_extension
-
-    if not accept:
-        log.warning('Could not determine accepted response type')
-        raise exc.exception_response(400)
+    try:
+        policy = ContentNegPolicy(config.content_negotiation_policy)
+    except ValueError:
+        log.debug(
+            f'Invalid value for config.content_negotiation_policy: {config.content_negotiation_policy}, '
+            f'defaulting to "extension"'
+        )
+        policy = ContentNegPolicy.extension
 
-    q: Optional[str]
-    if pfx and new_path:
-        q = f'{{{pfx}}}{new_path}'
-        new_path = f'/{alias}/{new_path}'
-    else:
-        q = new_path
+    accept, new_path, q = _process_content_negotiate(policy, alias, path, pfx, request)
 
     try:
-        accepter = MediaAccept(accept)
         for p in request.registry.plumbings:
-            state = {
-                entry: True,
-                'headers': {'Content-Type': None},
-                'accept': accepter,
-                'url': request.current_route_url(),
-                'select': q,
-                'match': match.lower() if match else match,
-                'path': new_path,
-                'stats': {},
-            }
+            state = PipeState(
+                entry_name=entry,
+                headers={'Content-Type': None},
+                accept=accept,
+                url=request.current_route_url(),
+                select=q,
+                match=match.lower() if match else match,
+                path=new_path,
+                stats={},
+            )
 
             r = p.process(request.registry.md, state=state, raise_exceptions=True, scheduler=request.registry.scheduler)
             log.debug(f'Plumbing process result: {r}')
             if r is None:
                 r = []
 
             response = Response()
-            _headers = state.get('headers', {})
-            response.headers.update(_headers)
-            ctype = _headers.get('Content-Type', None)
+            response.headers.update(state.headers)
+            ctype = state.headers.get('Content-Type', None)
             if not ctype:
-                r, t = _fmt(r, accepter)
+                r, t = _fmt(r, accept)
                 ctype = t
 
             response.text = b2u(r)
             response.size = len(r)
             response.content_type = ctype
-            cache_ttl = int(state.get('cache', 0))
-            response.expires = datetime.now() + timedelta(seconds=cache_ttl)
+            response.expires = datetime.now() + timedelta(seconds=state.cache)
             return response
     except ResourceException as ex:
         import traceback

src/pyff/builtins.py

@@ -25,7 +25,7 @@
 from pyff.decorators import deprecated
 from pyff.exceptions import MetadataException
 from pyff.logs import get_log
-from pyff.pipes import PipeException, PipelineCallback, Plumbing, pipe, registry
+from pyff.pipes import PipeException, PipeState, PipelineCallback, Plumbing, pipe, registry
 from pyff.samlmd import (
     annotate_entity,
     discojson_t,
@@ -383,14 +383,13 @@ def when(req: Plumbing.Request, condition: str, *values):
     The condition operates on the state: if 'foo' is present in the state (with any value), then the something branch is
     followed. If 'bar' is present in the state with the value 'bill' then the other branch is followed.
     """
-    c = req.state.get(condition, None)
-    if c is None:
+    if req.state.entry_name is None:
         log.debug(f'Condition {repr(condition)} not present in state {req.state}')
-    if c is not None and (not values or _any(values, c)):
+    if req.state.entry_name is not None and (not values or _any(values, req.state.entry_name)):
         if not isinstance(req.args, list):
             raise ValueError('Non-list arguments to "when" not allowed')
 
-        return Plumbing(pipeline=req.args, pid="%s.when" % req.plumbing.id).iprocess(req)
+        return Plumbing(pipeline=req.args, pid=f'{req.plumbing.id}.when').iprocess(req)
     return req.t
 
 
@@ -768,9 +767,9 @@ def select(req: Plumbing.Request, *opts):
 
     entities = resolve_entities(args, lookup_fn=req.md.store.select)
 
-    if req.state.get('match', None):  # TODO - allow this to be passed in via normal arguments
+    if req.state.match:  # TODO - allow this to be passed in via normal arguments
 
-        match = req.state['match']
+        match = req.state.match
 
         if isinstance(match, six.string_types):
             query = [match.lower()]
@@ -1435,11 +1434,11 @@ def emit(req: Plumbing.Request, ctype="application/xml", *opts):
         if not isinstance(d, six.binary_type):
             d = d.encode("utf-8")
         m.update(d)
-        req.state['headers']['ETag'] = m.hexdigest()
+        req.state.headers['ETag'] = m.hexdigest()
     else:
         raise PipeException("Empty")
 
-    req.state['headers']['Content-Type'] = ctype
+    req.state.headers['Content-Type'] = ctype
     if six.PY2:
         d = six.u(d)
     return d
@@ -1517,7 +1516,7 @@ def finalize(req: Plumbing.Request, *opts):
         if name is None or 0 == len(name):
             name = req.args.get('Name', None)
         if name is None or 0 == len(name):
-            name = req.state.get('url', None)
+            name = req.state.url
             if name and 'baseURL' in req.args:
 
                 try:
@@ -1569,7 +1568,7 @@ def finalize(req: Plumbing.Request, *opts):
         # TODO: offset can be None here, if validUntil is not a valid duration or ISO date
         #       What is the right action to take then?
         if offset:
-            req.state['cache'] = int(total_seconds(offset) / 50)
+            req.state.cache = int(total_seconds(offset) / 50)
 
     cache_duration = req.args.get('cacheDuration', e.get('cacheDuration', None))
     if cache_duration is not None and len(cache_duration) > 0:
@@ -1578,7 +1577,7 @@ def finalize(req: Plumbing.Request, *opts):
             raise PipeException("Unable to parse %s as xs:duration" % cache_duration)
 
         e.set('cacheDuration', cache_duration)
-        req.state['cache'] = int(total_seconds(offset))
+        req.state.cache = int(total_seconds(offset))
 
     return req.t