python3 -m pyupgrade --py39-plus on all python files. Removed six as a direct dependency.

Author: mikaelfrykholm

requirements.txt

@@ -18,7 +18,6 @@ requests
 requests_cache
 requests_file
 simplejson >=2.6.2
-six>=1.11.0
 whoosh
 wsgi-intercept
 xmldiff

src/pyff/api.py

@@ -2,7 +2,8 @@
 import threading
 from datetime import datetime, timedelta
 from json import dumps
-from typing import Any, Dict, Generator, Iterable, List, Mapping, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
+from collections.abc import Generator, Iterable, Mapping
 
 import pyramid.httpexceptions as exc
 import pytz
@@ -15,7 +16,7 @@
 from pyramid.request import Request
 from pyramid.response import Response
 from six import b
-from six.moves.urllib_parse import quote_plus
+from urllib.parse import quote_plus
 
 from pyff.constants import config
 from pyff.exceptions import ResourceException
@@ -30,7 +31,7 @@
 log = get_log(__name__)
 
 
-class NoCache(object):
+class NoCache:
     """Dummy implementation for when caching isn't enabled"""
 
     def __init__(self) -> None:
@@ -82,7 +83,7 @@ def status_handler(request: Request) -> Response:
     return response
 
 
-class MediaAccept(object):
+class MediaAccept:
     def __init__(self, accept: str):
         self._type = AcceptableType(accept)
 
@@ -110,7 +111,7 @@ def _is_xml(data: Any) -> bool:
     return isinstance(data, (etree._Element, etree._ElementTree))
 
 
-def _fmt(data: Any, accepter: MediaAccept) -> Tuple[str, str]:
+def _fmt(data: Any, accepter: MediaAccept) -> tuple[str, str]:
     """
     Format data according to the accepted content type of the requester.
     Return data as string (either XML or json) and a content-type.
@@ -162,7 +163,7 @@ def process_handler(request: Request) -> Response:
     """
     _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 _d(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()
@@ -195,7 +196,7 @@ def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional
 
     # Enable matching on scope.
     match = match.split('@').pop() if match and not match.endswith('@') else match
-    log.debug("match={}".format(match))
+    log.debug(f"match={match}")
 
     if not path_elem:
         path_elem = ['entities']
@@ -213,7 +214,7 @@ def _d(x: Optional[str], do_split: bool = True) -> Tuple[Optional[str], Optional
     if 'entities' not in alias:
         pfx = request.registry.aliases.get(alias, None)
         if pfx is None:
-            log.debug("alias {} not found - passing to storage lookup".format(alias))
+            log.debug(f"alias {alias} not found - passing to storage lookup")
             path = alias  # treat as path
 
     # content_negotiation_policy is one of three values:
@@ -353,11 +354,11 @@ def webfinger_handler(request: Request) -> Response:
     if resource is None:
         resource = request.host_url
 
-    jrd: Dict[str, Any] = dict()
+    jrd: dict[str, Any] = dict()
     dt = datetime.now() + timedelta(hours=1)
     jrd['expires'] = dt.isoformat()
     jrd['subject'] = request.host_url
-    links: List[Dict[str, Any]] = list()
+    links: list[dict[str, Any]] = list()
     jrd['links'] = links
 
     _dflt_rels = {
@@ -377,7 +378,7 @@ def _links(url: str, title: Any = None) -> None:
             suffix = ""
             if not url.endswith('/'):
                 suffix = _dflt_rels[r][0]
-            links.append(dict(rel=r, type=_dflt_rels[r][1], href='%s/%s%s' % (request.host_url, url, suffix)))
+            links.append(dict(rel=r, type=_dflt_rels[r][1], href=f'{request.host_url}/{url}{suffix}'))
 
     _links('/entities/')
     for a in request.registry.md.store.collections():
@@ -391,7 +392,7 @@ def _links(url: str, title: Any = None) -> None:
     aliases = request.registry.aliases
     for a in aliases.keys():
         for v in request.registry.md.store.attribute(aliases[a]):
-            _links('%s/%s' % (a, quote_plus(v)))
+            _links(f'{a}/{quote_plus(v)}')
 
     response = Response(dumps(jrd, default=json_serializer))
     response.headers['Content-Type'] = 'application/json'
@@ -407,7 +408,7 @@ def resources_handler(request: Request) -> Response:
     :return: a JSON representation of the set of resources currently loaded by the server
     """
 
-    def _infos(resources: Iterable[Resource]) -> List[Mapping[str, Any]]:
+    def _infos(resources: Iterable[Resource]) -> list[Mapping[str, Any]]:
         return [_info(r) for r in resources if r.info.state is not None]
 
     def _info(r: Resource) -> Mapping[str, Any]:
@@ -453,19 +454,19 @@ def search_handler(request: Request) -> Response:
     match = match.split('@').pop() if match and not match.endswith('@') else match
 
     entity_filter = request.params.get('entity_filter', '{http://pyff.io/role}idp')
-    log.debug("match={}".format(match))
+    log.debug(f"match={match}")
     store = request.registry.md.store
 
     def _response() -> Generator[bytes, bytes, None]:
-        yield b('[')
+        yield b'['
         in_loop = False
         entities = store.search(query=match.lower(), entity_filter=entity_filter)
         for e in entities:
             if in_loop:
-                yield b(',')
+                yield b','
             yield b(dumps(e))
             in_loop = True
-        yield b(']')
+        yield b']'
 
     response = Response(content_type='application/json')
     response.app_iter = _response()

src/pyff/builtins.py

@@ -17,11 +17,10 @@
 from typing import Dict, Optional
 
 import ipaddress
-import six
 import xmlsec
 from lxml import etree
 from lxml.etree import DocumentInvalid
-from six.moves.urllib_parse import quote_plus, urlparse
+from urllib.parse import quote_plus, urlparse
 
 from pyff.constants import NS
 from pyff.decorators import deprecated
@@ -104,7 +103,7 @@ def _map(req: Plumbing.Request, *opts):
 
     def _p(e):
         entity_id = e.get('entityID')
-        ip = Plumbing(pipeline=req.args, pid="{}.each[{}]".format(req.plumbing.pid, entity_id))
+        ip = Plumbing(pipeline=req.args, pid=f"{req.plumbing.pid}.each[{entity_id}]")
         ireq = Plumbing.Request(ip, req.md, t=e, scheduler=req.scheduler)
         ireq.set_id(entity_id)
         ireq.set_parent(req)
@@ -114,7 +113,7 @@ def _p(e):
 
     pool = ThreadPool()
     result = pool.map(_p, iter_entities(req.t), chunksize=10)
-    log.info("processed {} entities".format(len(result)))
+    log.info(f"processed {len(result)} entities")
 
 
 @pipe(name="then")
@@ -461,7 +460,7 @@ def sort(req: Plumbing.Request, *opts):
     if req.t is None:
         raise PipeException("Unable to sort empty document.")
 
-    _opts: Dict[str, Optional[str]] = dict(list(zip(opts[0:1], [" ".join(opts[1:])])))
+    _opts: dict[str, Optional[str]] = dict(list(zip(opts[0:1], [" ".join(opts[1:])])))
     if 'order_by' not in _opts:
         _opts['order_by'] = None
     sort_entities(req.t, _opts['order_by'])
@@ -837,7 +836,7 @@ def select(req: Plumbing.Request, *opts):
     if req.state.get('match', None):  # TODO - allow this to be passed in via normal arguments
         match = req.state['match']
 
-        if isinstance(match, six.string_types):
+        if isinstance(match, str):
             query = [match.lower()]
 
         def _strings(elt):
@@ -870,22 +869,22 @@ def _match(q, elt):
 
             if q is not None and len(q) > 0:
                 tokens = _strings(elt)
-                p = re.compile(r'\b{}'.format(q), re.IGNORECASE)
+                p = re.compile(fr'\b{q}', re.IGNORECASE)
                 for tstr in tokens:
                     if p.search(tstr):
                         return tstr
             return None
 
-        log.debug("matching {} in {} entities".format(match, len(entities)))
+        log.debug(f"matching {match} in {len(entities)} entities")
         entities = list(filter(lambda e: _match(match, e) is not None, entities))
-        log.debug("returning {} entities after match".format(len(entities)))
+        log.debug(f"returning {len(entities)} entities after match")
 
     ot = entitiesdescriptor(entities, name)
     if ot is None:
         raise PipeException("empty select - stop")
 
     if req.plumbing.id != name:
-        log.debug("storing synthetic collection {}".format(name))
+        log.debug(f"storing synthetic collection {name}")
         req.store.update(ot, name)
 
     return ot
@@ -1202,7 +1201,7 @@ def stats(req: Plumbing.Request, *opts):
         raise PipeException("Your pipeline is missing a select statement.")
 
     print("---")
-    print("total size:     {:d}".format(req.store.size()))
+    print(f"total size:     {req.store.size():d}")
     if not hasattr(req.t, 'xpath'):
         raise PipeException("Unable to call stats on non-XML")
 
@@ -1303,7 +1302,7 @@ def xslt(req: Plumbing.Request, *opts):
     if stylesheet is None:
         raise PipeException("xslt requires stylesheet")
 
-    params = dict((k, "'%s'" % v) for (k, v) in list(req.args.items()))
+    params = {k: "'%s'" % v for (k, v) in list(req.args.items())}
     del params['stylesheet']
     try:
         return root(xslt_transform(req.t, stylesheet, params))
@@ -1429,13 +1428,13 @@ def check_xml_namespaces(req: Plumbing.Request, *opts):
         raise PipeException("Your pipeline is missing a select statement.")
 
     def _verify(elt):
-        if isinstance(elt.tag, six.string_types):
+        if isinstance(elt.tag, str):
             for prefix, uri in list(elt.nsmap.items()):
                 if not uri.startswith('urn:'):
                     u = urlparse(uri)
                     if u.scheme not in ('http', 'https'):
                         raise MetadataException(
-                            "Namespace URIs must be be http(s) URIs ('{}' declared on {})".format(uri, elt.tag)
+                            f"Namespace URIs must be be http(s) URIs ('{uri}' declared on {elt.tag})"
                         )
 
     with_tree(root(req.t), _verify)
@@ -1508,7 +1507,7 @@ def certreport(req: Plumbing.Request, *opts):
     error_bits = int(req.args.get('error_bits', "1024"))
     warning_bits = int(req.args.get('warning_bits', "2048"))
 
-    seen: Dict[str, bool] = {}
+    seen: dict[str, bool] = {}
     for eid in req.t.xpath("//md:EntityDescriptor/@entityID", namespaces=NS, smart_strings=False):
         for cd in req.t.xpath(
             "md:EntityDescriptor[@entityID='%s']//ds:X509Certificate" % eid, namespaces=NS, smart_strings=False
@@ -1530,17 +1529,17 @@ def certreport(req: Plumbing.Request, *opts):
                             entity_elt,
                             "certificate-error",
                             "keysize too small",
-                            "%s has keysize of %s bits (less than %s)" % (cert.getSubject(), keysize, error_bits),
+                            "{} has keysize of {} bits (less than {})".format(cert.getSubject(), keysize, error_bits),
                         )
-                        log.error("%s has keysize of %s" % (eid, keysize))
+                        log.error("{} has keysize of {}".format(eid, keysize))
                     elif keysize < warning_bits:
                         annotate_entity(
                             entity_elt,
                             "certificate-warning",
                             "keysize small",
-                            "%s has keysize of %s bits (less than %s)" % (cert.getSubject(), keysize, warning_bits),
+                            "{} has keysize of {} bits (less than {})".format(cert.getSubject(), keysize, warning_bits),
                         )
-                        log.warning("%s has keysize of %s" % (eid, keysize))
+                        log.warning("{} has keysize of {}".format(eid, keysize))
 
                     notafter = cert.getNotAfter()
                     if notafter is None:
@@ -1560,23 +1559,23 @@ def certreport(req: Plumbing.Request, *opts):
                                     entity_elt,
                                     "certificate-error",
                                     "certificate has expired",
-                                    "%s expired %s ago" % (cert.getSubject(), -dt),
+                                    "{} expired {} ago".format(cert.getSubject(), -dt),
                                 )
-                                log.error("%s expired %s ago" % (eid, -dt))
+                                log.error("{} expired {} ago".format(eid, -dt))
                             elif total_seconds(dt) < warning_seconds:
                                 annotate_entity(
                                     entity_elt,
                                     "certificate-warning",
                                     "certificate about to expire",
-                                    "%s expires in %s" % (cert.getSubject(), dt),
+                                    "{} expires in {}".format(cert.getSubject(), dt),
                                 )
-                                log.warning("%s expires in %s" % (eid, dt))
+                                log.warning("{} expires in {}".format(eid, dt))
                         except ValueError as ex:
                             annotate_entity(
                                 entity_elt,
                                 "certificate-error",
                                 "certificate has unknown expiration time",
-                                "%s unknown expiration time %s" % (cert.getSubject(), notafter),
+                                "{} unknown expiration time {}".format(cert.getSubject(), notafter),
                             )
 
                     req.store.update(entity_elt)
@@ -1623,7 +1622,7 @@ def emit(req: Plumbing.Request, ctype="application/xml", *opts):
 
     if d is not None:
         m = hashlib.sha1()
-        if not isinstance(d, six.binary_type):
+        if not isinstance(d, bytes):
             d = d.encode("utf-8")
         m.update(d)
         req.state['headers']['ETag'] = m.hexdigest()
@@ -1632,7 +1631,7 @@ def emit(req: Plumbing.Request, ctype="application/xml", *opts):
 
     req.state['headers']['Content-Type'] = ctype
     if six.PY2:
-        d = six.u(d)
+        d = d
     return d
 
 
@@ -1664,7 +1663,7 @@ def signcerts(req: Plumbing.Request, *opts):
 
 @pipe
 def finalize(req: Plumbing.Request, *opts):
-    """
+    r"""
     Prepares the working document for publication/rendering.
 
     :param req: The request
@@ -1716,7 +1715,7 @@ def finalize(req: Plumbing.Request, *opts):
                     # TODO: Investigate this error, which is probably correct:
                     #       error: On Python 3 '{}'.format(b'abc') produces "b'abc'", not 'abc';
                     #       use '{!r}'.format(b'abc') if this is desired behavior
-                    name = "{}://{}{}".format(base_url.scheme, base_url.netloc, name_url.path)  # type: ignore
+                    name = f"{base_url.scheme}://{base_url.netloc}{name_url.path}"  # type: ignore
                     log.debug("-------- using Name: %s" % name)
                 except ValueError as ex:
                     log.debug(f'Got an exception while finalizing: {ex}')
@@ -1752,7 +1751,7 @@ def finalize(req: Plumbing.Request, *opts):
                 offset = dt - now
                 e.set('validUntil', datetime2iso(dt))
             except ValueError as ex:
-                log.error("Unable to parse validUntil: %s (%s)" % (valid_until, ex))
+                log.error("Unable to parse validUntil: {} ({})".format(valid_until, ex))
 
         # set a reasonable default: 50% of the validity
         # we replace this below if we have cacheDuration set