Server : LiteSpeed System : Linux in-mum-web1112.main-hosting.eu 4.18.0-553.34.1.lve.el8.x86_64 #1 SMP Thu Jan 9 16:30:32 UTC 2025 x86_64 User : u451330669 ( 451330669) PHP Version : 8.2.27 Disable Function : NONE Directory :  /opt/alt/python311/lib/python3.11/site-packages/pyroute2/requests/

Upload File : current_dir [ Writeable ] document_root [ Writeable ]

'''
General request and RTNL object data filters.
'''

import weakref
from collections import ChainMap


class RequestProcessor(dict):
    def __init__(self, field_filter=None, context=None, prime=None):
        self.field_filter = field_filter
        self.context = (
            context if isinstance(context, (dict, weakref.ProxyType)) else {}
        )
        self.combined = ChainMap(self, self.context)
        if isinstance(prime, dict):
            self.update(prime)

    def __setitem__(self, key, value):
        if value is None:
            return
        if key in self:
            del self[key]
        for nkey, nvalue in self.filter(key, value).items():
            super(RequestProcessor, self).__setitem__(nkey, nvalue)

    def filter(self, key, value):
        if hasattr(self.field_filter, '_key_transform'):
            key = self.field_filter._key_transform(key)
        if (
            hasattr(self.field_filter, '_allowed')
            and key not in self.field_filter._allowed
        ):
            return {}
        if hasattr(
            self.field_filter, 'policy'
        ) and not self.field_filter.policy(key):
            return {}
        return getattr(
            self.field_filter, f'set_{key}', lambda *argv: {key: value}
        )(self.combined, value)

    def update(self, prime):
        for key, value in tuple(prime.items()):
            self[key] = value

    def set_filter(self, field_filter):
        self.field_filter = field_filter
        return self

    def apply_filter(self, field_filter):
        self.field_filter = field_filter
        self.update(self)
        return self

    def finalize(self, cmd_context=None):
        if hasattr(self.field_filter, 'finalize_for_iproute'):
            # old interface
            self.field_filter.finalize_for_iproute(self.combined, cmd_context)
        if hasattr(self.field_filter, 'finalize'):
            # new interface
            self.field_filter.finalize(self.combined)
        return self