Fix Pydantic deprecation copy > model_copy

Author: mikaelfrykholm

src/pyff/builtins.py

@@ -256,7 +256,7 @@ def fork(req: Plumbing.Request, *opts):
     **parsecopy**
 
     Due to a hard to find bug, fork which uses deepcopy can lose some namespaces. The parsecopy argument is a workaround.
-    It uses a brute force serialisation and deserialisation to get around the bug. 
+    It uses a brute force serialisation and deserialisation to get around the bug.
 
     .. code-block:: yaml
 
@@ -676,7 +676,7 @@ def load(req: Plumbing.Request, *opts):
         url = r.pop(0)
 
         # Copy parent node opts as a starting point
-        child_opts = req.md.rm.opts.copy(update={"via": [], "cleanup": [], "verify": None, "alias": url})
+        child_opts = req.md.rm.opts.model_copy(update={"via": [], "cleanup": [], "verify": None, "alias": url})
 
         while len(r) > 0:
             elt = r.pop(0)
@@ -702,7 +702,7 @@ def load(req: Plumbing.Request, *opts):
                 child_opts.verify = elt
 
         # override anything in child_opts with what is in opts
-        child_opts = child_opts.copy(update=_opts)
+        child_opts = child_opts.model_copy(update=_opts)
 
         req.md.rm.add_child(url, child_opts)
 
@@ -814,7 +814,7 @@ def select(req: Plumbing.Request, *opts):
             else:
                 _opts['as'] = opts[i]
                 if i + 1 < len(opts):
-                    more_opts = opts[i + 1:]
+                    more_opts = opts[i + 1 :]
                     _opts.update(dict(list(zip(more_opts[::2], more_opts[1::2]))))
                     break
 
@@ -835,7 +835,6 @@ def select(req: Plumbing.Request, *opts):
     entities = resolve_entities(args, lookup_fn=req.md.store.select, dedup=dedup)
 
     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):
@@ -1304,14 +1303,15 @@ 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 = dict((k, "'%s'" % v) for (k, v) in list(req.args.items()))
     del params['stylesheet']
     try:
         return root(xslt_transform(req.t, stylesheet, params))
     except Exception as ex:
         log.debug(traceback.format_exc())
         raise ex
 
+
 @pipe
 def indent(req: Plumbing.Request, *opts):
     """
@@ -1710,7 +1710,6 @@ def finalize(req: Plumbing.Request, *opts):
         if name is None or 0 == len(name):
             name = req.state.get('url', None)
             if name and 'baseURL' in req.args:
-
                 try:
                     name_url = urlparse(name)
                     base_url = urlparse(req.args.get('baseURL'))

src/pyff/parse.py

@@ -9,7 +9,7 @@
 
 from pyff.constants import NS
 from pyff.logs import get_log
-from pyff.resource import Resource,ResourceInfo
+from pyff.resource import Resource, ResourceInfo
 from pyff.utils import find_matching_files, parse_xml, root, unicode_stream, utc_now
 
 __author__ = 'leifj'
@@ -30,8 +30,10 @@ def _format_key(k: str) -> str:
         res = {_format_key(k): v for k, v in self.dict().items()}
         return res
 
+
 ResourceInfo.model_rebuild()
 
+
 class ParserException(Exception):
     def __init__(self, msg, wrapped=None, data=None):
         self._wraped = wrapped
@@ -84,7 +86,7 @@ def parse(self, resource: Resource, content: str) -> ParserInfo:
         info = ParserInfo(description='Directory', expiration_time='never expires')
         n = 0
         for fn in find_matching_files(content, self.extensions):
-            child_opts = resource.opts.copy(update={'alias': None})
+            child_opts = resource.opts.model_copy(update={'alias': None})
             resource.add_child("file://" + urlescape(fn), child_opts)
             n += 1
 
@@ -122,7 +124,7 @@ def parse(self, resource: Resource, content: str) -> ParserInfo:
                 if len(fingerprints) > 0:
                     fp = fingerprints[0]
                 log.debug("XRD: {} verified by {}".format(link_href, fp))
-                child_opts = resource.opts.copy(update={'alias': None})
+                child_opts = resource.opts.model_copy(update={'alias': None})
                 resource.add_child(link_href, child_opts)
         resource.last_seen = utc_now().replace(microsecond=0)
         resource.expire_time = None

src/pyff/samlmd.py

@@ -86,7 +86,10 @@ def find_merge_strategy(strategy_name):
 
 
 def parse_saml_metadata(
-    source: BytesIO, opts: ResourceOpts, base_url=None, validation_errors: Optional[Dict[str, Any]] = None,
+    source: BytesIO,
+    opts: ResourceOpts,
+    base_url=None,
+    validation_errors: Optional[Dict[str, Any]] = None,
 ):
     """Parse a piece of XML and return an EntitiesDescriptor element after validation.
 
@@ -192,7 +195,10 @@ def _extra_md(_t, info, **kwargs):
             location = kwargs.get('location')
             sp_entity = sp_entities.find("{%s}EntityDescriptor[@entityID='%s']" % (NS['md'], entityID))
             if sp_entity is not None:
-                md_source = sp_entity.find("{%s}SPSSODescriptor/{%s}Extensions/{%s}TrustInfo/{%s}MetadataSource[@src='%s']" % (NS['md'], NS['md'], NS['ti'], NS['ti'], location))
+                md_source = sp_entity.find(
+                    "{%s}SPSSODescriptor/{%s}Extensions/{%s}TrustInfo/{%s}MetadataSource[@src='%s']"
+                    % (NS['md'], NS['md'], NS['ti'], NS['ti'], location)
+                )
                 for e in iter_entities(_t):
                     md_source.append(e)
             return etree.Element("{%s}EntitiesDescriptor" % NS['md'])
@@ -205,11 +211,14 @@ def _extra_md(_t, info, **kwargs):
                 entityID = e.get('entityID')
                 info.entities.append(entityID)
 
-                md_source = e.find("{%s}SPSSODescriptor/{%s}Extensions/{%s}TrustInfo/{%s}MetadataSource" % (NS['md'], NS['md'], NS['ti'], NS['ti']))
+                md_source = e.find(
+                    "{%s}SPSSODescriptor/{%s}Extensions/{%s}TrustInfo/{%s}MetadataSource"
+                    % (NS['md'], NS['md'], NS['ti'], NS['ti'])
+                )
                 if md_source is not None:
                     location = md_source.attrib.get('src')
                     if location is not None:
-                        child_opts = resource.opts.copy(update={'alias': entityID})
+                        child_opts = resource.opts.model_copy(update={'alias': entityID})
                         r = resource.add_child(location, child_opts)
                         kwargs = {
                             'entityID': entityID,
@@ -311,7 +320,7 @@ def parse(self, resource: Resource, content: str) -> EidasMDParserInfo:
                                 info.scheme_territory, location, fp, args.get('country_code')
                             )
                         )
-                        child_opts = resource.opts.copy(update={'alias': None})
+                        child_opts = resource.opts.model_copy(update={'alias': None})
                         child_opts.verify = fp
                         r = resource.add_child(location, child_opts)
 
@@ -725,7 +734,6 @@ def entity_domains(entity):
 
 
 def entity_extended_display_i18n(entity, default_lang=None):
-
     name_dict = lang_dict(entity.iter("{%s}OrganizationName" % NS['md']), lambda e: e.text, default_lang=default_lang)
     name_dict.update(
         lang_dict(entity.iter("{%s}OrganizationDisplayName" % NS['md']), lambda e: e.text, default_lang=default_lang)
@@ -981,7 +989,9 @@ def discojson_sp(e, global_trust_info=None, global_md_sources=None):
 
     sp['entityID'] = e.get('entityID', None)
 
-    md_sources = e.findall("{%s}SPSSODescriptor/{%s}Extensions/{%s}TrustInfo/{%s}MetadataSource" % (NS['md'], NS['md'], NS['ti'], NS['ti']))
+    md_sources = e.findall(
+        "{%s}SPSSODescriptor/{%s}Extensions/{%s}TrustInfo/{%s}MetadataSource" % (NS['md'], NS['md'], NS['ti'], NS['ti'])
+    )
 
     sp['extra_md'] = {}
     for md_source in md_sources:
@@ -1041,7 +1051,6 @@ def discojson_sp(e, global_trust_info=None, global_md_sources=None):
 
 
 def discojson_sp_attr(e):
-
     attribute = "https://refeds.org/entity-selection-profile"
     b64_trustinfos = entity_attribute(e, attribute)
     if b64_trustinfos is None:
@@ -1395,7 +1404,7 @@ def get_key(e):
                 except AttributeError:
                     pass
             except IndexError:
-                log.warning("Sort pipe: unable to sort entity by '%s'. " "Entity '%s' has no such value" % (sxp, eid))
+                log.warning("Sort pipe: unable to sort entity by '%s'. Entity '%s' has no such value" % (sxp, eid))
         except TypeError:
             pass