Source code for woob.capabilities.base

# Copyright(C) 2010-2013 Christophe Benz, Romain Bignon, Julien Hebert
#
# This file is part of woob.
#
# woob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# woob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with woob. If not, see <http://www.gnu.org/licenses/>.

from __future__ import annotations

from typing import Dict, Iterable, Type, Any, Tuple, TypeVar, overload

from collections import OrderedDict, deque
import warnings
import re
from decimal import Decimal
from copy import deepcopy, copy

from woob.tools.misc import to_unicode


__all__ = [
    'UserError', 'FieldNotFound', 'NotAvailable', 'FetchError', 'NotLoaded',
    'Capability', 'Field', 'IntField', 'DecimalField', 'FloatField',
    'StringField', 'BytesField', 'BoolField', 'Enum', 'EnumField', 'empty',
    'BaseObject', 'find_object', 'find_object_any_match', 'strict_find_object',
    'capability_to_string',
]


class EnumMeta(type):
    @classmethod
    def __prepare__(mcs, name, bases, **kwargs):
        # in python3.6, default namespace keeps declaration order
        # in python>=3 but <3.6, force ordered namespace
        # doesn't work in python2
        return OrderedDict()

    def __init__(cls, name, bases, attrs, *args, **kwargs):
        super(EnumMeta, cls).__init__(name, bases, attrs, *args, **kwargs)
        attrs = [(k, v) for k, v in attrs.items() if not callable(v) and not k.startswith('__')]
        cls.__members__ = OrderedDict(attrs)

    def __setattr__(cls, name, value):
        super(EnumMeta, cls).__setattr__(name, value)
        if not callable(value) and not name.startswith('__'):
            cls.__members__[name] = value

    def __call__(cls, *args, **kwargs):
        raise ValueError("Enum type can't be instanciated")

    @property
    def _items(cls):
        return cls.__members__.items()

    @property
    def _keys(cls):
        return cls.__members__.keys()

    @property
    def _values(cls):
        return cls.__members__.values()

    @property
    def _types(cls):
        return set(map(type, cls._values))

    def __iter__(cls):
        return iter(cls.__members__.values())

    def __len__(cls):
        return len(cls.__members__)

    def __contains__(cls, value):
        return value in cls.__members__.values()

    def __getitem__(cls, k):
        return cls.__members__[k]


[docs]class Enum(metaclass=EnumMeta): pass
[docs]def empty(value: Any) -> bool: """ Checks if a value is empty (None, NotLoaded or NotAvailable). :rtype: :class:`bool` """ return value is None or isinstance(value, EmptyType)
T = TypeVar('T') @overload def find_object( mylist: Iterable[T], error: None = None, **kwargs ) -> T | None: ... @overload def find_object( mylist: Iterable[T], error: Type[Exception], **kwargs ) -> T: ...
[docs]def find_object( mylist, error = None, **kwargs ): """ Very simple tools to return an object with the matching parameters in kwargs. """ for a in mylist: for key, value in kwargs.items(): if getattr(a, key) != value: break else: return a if error is not None: raise error() return None
@overload def find_object_any_match( objects: Iterable[T], key_value_pairs: Iterable[Tuple[str, Any]], error: None = None, with_priority: bool = True, ignore_empty: bool = True ) -> T | None: ... @overload def find_object_any_match( objects: Iterable[T], key_value_pairs: Iterable[Tuple[str, Any]], error: Type[Exception], with_priority: bool = True, ignore_empty: bool = True ) -> T: ...
[docs]def find_object_any_match( objects: Iterable[T], key_value_pairs: Iterable[Tuple[str, Any]], error: Type[Exception] | None = None, with_priority: bool = True, ignore_empty: bool = True ) -> T | None: """ Tool method that returns an object that match any parameter given in key_value_pairs. Compared with find_object, this method does not need all parameters to be matching. If no priority is set, the method will return the first successful match. With several key-values set, this may not be the object instance with the most key matches. :param objects: The list or iterator of objects that can match :type objectlist: list :param key_value_pairs: The key-values that the object should have to match. This is given as a list of pairs, where each pair represent a key-value, the first item being the key, and the second is the value. :type key_value_pairs: list :param error: The error to raise in case no object matches. :type error: Exception or None :param with_priority: If True, the key-value pairs order decides the priority of the matching. With priority enabled, an object matching with a more important key will be matched over another object matching with a less important keys. This means that all objects will be iterated in order to assure this priority, unless the most preferred key matches, in which case the iteration stops early. This feature is enabled by default. See the example sections for more details. :type with_priority: bool :param ignore_empty: If True, empty key-values (None, EmptyType) will not be matched. True by default. :type ignore_empty: bool :return: The matching object if any Examples: >>> class Point: ... def __init__(self, x, y): ... self.x = x ... self.y = y ... def __repr__(self): ... return f'Point({self.x},{self.y})' >>> points = (Point(0,0), Point(1, 2), Point(None, 3)) >>> find_object_any_match(points, (('x', 0), ('y', 1))) # Will match on x with first point Point(0,0) >>> find_object_any_match(points, (('x', 2), ('y', 2))) # Will match on y with second point Point(1,2) >>> find_object_any_match(points, (('x', 1), ('y', 0))) # Will match on x with second point because priority Point(1,2) >>> find_object_any_match(points, (('x', 1), ('y', 0)), with_priority=False) # Will return first successful match Point(0,0) >>> find_object_any_match(points, (('x', None),)) # second condition (None) is ignored by default None >>> find_object_any_match(points, (('x', None),), ignore_empty=False) # will match last point Point(None,3) >>> find_object_any_match(points, (('x', -1),), error=ValueError) # Will raise ValueError ValueError: """ # Will be used if with_priority is true found = None found_index = None for o in objects: for index, (key, value) in enumerate(key_value_pairs): # There is no point trying to match as it will be not priorized if found_index and found_index <= index: break o_value = getattr(o, key) if ( ignore_empty and (empty(value) or empty(o_value)) or o_value != value ): continue # First key is a special case with priority, we can return early # if we find a match on the first key # Also we only keep more priorized matches if not with_priority or index == 0: return o found = o found_index = index if not found and error is not None: raise error() return found
[docs]def strict_find_object(mylist: Iterable[BaseObject], error=None, **kwargs) -> BaseObject | None: """ Tools to return an object with the matching parameters in kwargs. Parameters with empty value are skipped """ kwargs = {k: v for k, v in kwargs.items() if not empty(v)} if kwargs: return find_object(mylist, error=error, **kwargs) if error is not None: raise error() return None
[docs]class UserError(Exception): """ Exception containing an error message for user. """
[docs]class FieldNotFound(Exception): """ A field isn't found. :param obj: object :type obj: :class:`BaseObject` :param field: field not found :type field: :class:`Field` """ def __init__(self, obj: BaseObject, field: str): super().__init__('Field "%s" not found for object %s' % (field, obj))
class ConversionWarning(UserWarning): """ A field's type was changed when setting it. Ideally, the module should use the right type before setting it. """ class AttributeCreationWarning(UserWarning): """ A non-field attribute has been created with a name not prefixed with a _. """ class EmptyType: """ Parent class for NotAvailableType, NotLoadedType and FetchErrorType. """ def __str__(self): return repr(self) def __copy__(self): return self def __deepcopy__(self, memo): return self def __nonzero__(self): return False __bool__ = __nonzero__ class NotAvailableType(EmptyType): """ NotAvailable is a constant to use on non available fields. """ def __repr__(self): return 'NotAvailable' def __str__(self): return 'Not available' NotAvailable = NotAvailableType() class NotLoadedType(EmptyType): """ NotLoaded is a constant to use on not loaded fields. When you use :func:`woob.tools.backend.Module.fillobj` on a object based on :class:`BaseObject`, it will request all fields with this value. """ def __repr__(self): return 'NotLoaded' def __str__(self): return 'Not loaded' NotLoaded = NotLoadedType() class FetchErrorType(EmptyType): """ FetchError is a constant to use when parsing a non-mandatory field raises an exception. """ def __repr__(self): return 'FetchError' def __str__(self): return 'Not mandatory' FetchError = FetchErrorType()
[docs]class Capability: """ This is the base class for all capabilities. A capability may define abstract methods (which raise :class:`NotImplementedError`) with an explicit docstring to tell backends how to implement them. Also, it may define some *objects*, using :class:`BaseObject`. """
[docs]class Field: """ Field of a :class:`BaseObject` class. :param doc: docstring of the field :type doc: :class:`str` :param args: list of types accepted :param default: default value of this field. If not specified, :class:`NotLoaded` is used. """ _creation_counter = 0 def __init__(self, doc, *args, **kwargs): self.types = () self.value = kwargs.get('default', NotLoaded) self.doc = doc self.mandatory = kwargs.get('mandatory', True) for arg in args: if isinstance(arg, type) or isinstance(arg, str): self.types += (arg,) else: raise TypeError('Arguments must be types or strings of type name') self._creation_counter = Field._creation_counter Field._creation_counter += 1
[docs] def convert(self, value): """ Convert value to the wanted one. """ return value
[docs]class IntField(Field): """ A field which accepts only :class:`int` types. """ def __init__(self, doc, **kwargs): super().__init__(doc, int, **kwargs)
[docs] def convert(self, value): return int(value)
[docs]class BoolField(Field): """ A field which accepts only :class:`bool` type. """ def __init__(self, doc, **kwargs): super().__init__(doc, bool, **kwargs)
[docs] def convert(self, value): return bool(value)
[docs]class DecimalField(Field): """ A field which accepts only :class:`decimal` type. """ def __init__(self, doc, **kwargs): super().__init__(doc, Decimal, **kwargs)
[docs] def convert(self, value): if isinstance(value, Decimal): return value return Decimal(value)
[docs]class FloatField(Field): """ A field which accepts only :class:`float` type. """ def __init__(self, doc, **kwargs): super(FloatField, self).__init__(doc, float, **kwargs)
[docs] def convert(self, value): return float(value)
[docs]class StringField(Field): """ A field which accepts only :class:`str` strings. """ def __init__(self, doc, **kwargs): super(StringField, self).__init__(doc, str, **kwargs)
[docs] def convert(self, value): return to_unicode(value)
[docs]class BytesField(Field): """ A field which accepts only :class:`bytes` strings. """ def __init__(self, doc, **kwargs): super(BytesField, self).__init__(doc, bytes, **kwargs)
[docs] def convert(self, value): if isinstance(value, str): value = value.encode('utf-8') return bytes(value)
[docs]class EnumField(Field): def __init__(self, doc, enum, **kwargs): if not issubclass(enum, Enum): raise TypeError('invalid enum type: %r' % enum) super(EnumField, self).__init__(doc, *enum._types, **kwargs) self.enum = enum
[docs] def convert(self, value): if value not in self.enum._values: raise ValueError('value %r does not belong to enum %s' % (value, self.enum)) return value
class _BaseObjectMeta(type): def __new__(cls, name, bases, attrs): fields = [(field_name, attrs.pop(field_name)) for field_name, obj in list(attrs.items()) if isinstance(obj, Field)] fields.sort(key=lambda x: x[1]._creation_counter) new_class = super(_BaseObjectMeta, cls).__new__(cls, name, bases, attrs) if new_class._fields is None: new_class._fields = OrderedDict() else: new_class._fields = deepcopy(new_class._fields) new_class._fields.update(fields) if new_class.__doc__ is None: new_class.__doc__ = '' for name, field in new_class._fields.items(): doc = '(%s) %s' % (', '.join([':class:`%s`' % v.__name__ if isinstance(v, type) else v for v in field.types]), field.doc) if field.value is not NotLoaded: doc += ' (default: %s)' % field.value new_class.__doc__ += '\n:var %s: %s' % (name, doc) return new_class
[docs]class BaseObject(metaclass=_BaseObjectMeta): """ This is the base class for a capability object. A capability interface may specify to return several kind of objects, to formalise retrieved information from websites. As python is a flexible language where variables are not typed, we use a system to force backends to set wanted values on all fields. To do that, we use the :class:`Field` class and all derived ones. For example:: class Transfer(BaseObject): " Transfer from an account to a recipient. " amount = DecimalField('Amount to transfer') date = Field('Date of transfer', str, date, datetime) origin = Field('Origin of transfer', int, str) recipient = Field('Recipient', int, str) The docstring is mandatory. """ id: str | None = None backend: str | None = None _fields: Dict[str, Field] = {} # XXX remove it? url = StringField('url') def __init__( self, id: str = '', url: str | NotLoadedType | NotAvailableType = NotLoaded, backend=None ): self.id = id or '' self.backend = backend self._fields = deepcopy(self._fields) self.__setattr__('url', url) @property def fullid(self) -> str: """ Full ID of the object, in form '**ID@backend**'. """ return '%s@%s' % (self.id, self.backend) def __iscomplete__(self) -> bool: """ Return True if the object is completed. It is useful when the object is a field of an other object which is going to be filled. The default behavior is to iter on fields (with iter_fields) and if a field is NotLoaded, return False. """ for key, value in self.iter_fields(): if value is NotLoaded: return False return True
[docs] def copy(self) -> BaseObject: obj = copy(self) obj._fields = copy(self._fields) for k in obj._fields: obj._fields[k] = copy(obj._fields[k]) return obj
def __deepcopy__(self, memo) -> BaseObject: return self.copy()
[docs] def set_empty_fields(self, value: Any, excepts=()): """ Set the same value on all empty fields. :param value: value to set on all empty fields :param excepts: if specified, do not change fields listed """ for key, old_value in self.iter_fields(): if empty(old_value) and key not in excepts: setattr(self, key, value)
[docs] def iter_fields(self) -> Iterable[Tuple[str, Any]]: """ Iterate on the fields keys and values. Can be overloaded to iterate on other things. :rtype: iter[(key, value)] """ if hasattr(self, 'id') and self.id is not None: yield 'id', self.id for name, field in self._fields.items(): yield name, field.value
def __eq__(self, obj) -> bool: if isinstance(obj, BaseObject): return self.backend == obj.backend and self.id == obj.id else: return False def __getattr__(self, name: str) -> Any: if self._fields is not None and name in self._fields: return self._fields[name].value else: raise AttributeError("'%s' object has no attribute '%s'" % ( self.__class__.__name__, name)) def __setattr__(self, name: str, value: Any): try: attr = (self._fields or {})[name] except KeyError: if name not in dir(self) and not name.startswith('_'): warnings.warn('Creating a non-field attribute %s. Please prefix it with _' % name, AttributeCreationWarning, stacklevel=2) object.__setattr__(self, name, value) else: if not empty(value): try: # Try to convert value to the wanted one. nvalue = attr.convert(value) except (TypeError, ValueError, ArithmeticError): # error during conversion, it will probably not # match the wanted following types, so we'll # raise ValueError. pass else: # If the value was converted if nvalue is not value: warnings.warn('Value %s was converted from %s to %s' % (name, type(value), type(nvalue)), ConversionWarning, stacklevel=2) value = nvalue actual_types = _resolve_types(attr.types) if not isinstance(value, actual_types) and not empty(value): raise ValueError( 'Value for "%s" needs to be of type %r, not %r' % ( name, actual_types, type(value))) attr.value = value def __delattr__(self, name: str): try: self._fields.pop(name) except KeyError: object.__delattr__(self, name)
[docs] def to_dict(self) -> Dict[str, Any]: def iter_decorate(d): for key, value in d: if key == 'id' and self.backend is not None: value = self.fullid yield key, value fields_iterator = self.iter_fields() return OrderedDict(iter_decorate(fields_iterator))
def __getstate__(self) -> Dict[str, Any]: d = self.to_dict() d.update((k, v) for k, v in self.__dict__.items() if k != '_fields') return d
[docs] @classmethod def from_dict(cls, values: Dict[str, Any], backend: str | None = None): self = cls() for attr in values: setattr(self, attr, values[attr]) return self
def __setstate__(self, state: Dict[str, Any]): self._fields = deepcopy(self._fields) # because yaml does not call __init__ for k in state: setattr(self, k, state[k]) def __dir__(self): return list(super(BaseObject, self).__dir__()) + list(self._fields.keys())
def _resolve_types(types): actual_types = () for v in types: if isinstance(v, str): # the following is a (almost) copy/paste from # https://stackoverflow.com/questions/11775460/lexical-cast-from-string-to-type q = deque([object]) while q: t = q.popleft() if t.__name__ == v: actual_types += (t,) else: try: # keep looking! q.extend(t.__subclasses__()) except TypeError: # type.__subclasses__ needs an argument for # whatever reason. if t is type: continue else: raise else: actual_types += (v,) return actual_types class Currency: CURRENCIES = OrderedDict([ ('EUR', ('€', 'EURO', 'EUROS')), ('CHF', ('CHF',)), ('USD', ('$', '$US')), ('GBP', ('£',)), ('LBP', ('ل.ل',)), ('AED', ('AED',)), ('XOF', ('XOF',)), ('RUB', ('руб',)), ('SGD', ('SGD',)), ('BRL', ('R$',)), ('MXN', ('$',)), ('JPY', ('¥',)), ('TRY', ('₺', 'TRY')), ('RON', ('lei',)), ('COP', ('$',)), ('NOK', ('kr',)), ('CNY', ('¥',)), ('RSD', ('din',)), ('ZAR', ('rand',)), ('MYR', ('RM',)), ('HUF', ('Ft',)), ('HKD', ('HK$',)), ('TWD', ('NT$',)), ('QAR', ('QR',)), ('MAD', ('MAD',)), ('ARS', ('ARS',)), ('AUD', ('AUD',)), ('CAD', ('CAD',)), ('NZD', ('NZD',)), ('BHD', ('BHD',)), ('SEK', ('SEK',)), ('DKK', ('DKK',)), ('LUF', ('LUF',)), ('KZT', ('KZT',)), ('PLN', ('PLN',)), ('ILS', ('ILS',)), ('THB', ('THB',)), ('INR', ('₹', 'INR')), ('PEN', ('S/',)), ('IDR', ('Rp',)), ('KWD', ('KD',)), ('KRW', ('₩',)), ('CZK', ('Kč',)), ('EGP', ('E£',)), ('ISK', ('Íkr', 'kr')), ('XPF', ('XPF',)), ('SAR', ('SAR',)), ('BGN', ('лв',)), ('HRK', ('HRK', 'kn')), ]) EXTRACTOR = re.compile(r'[()\d\s,\.\-]', re.UNICODE) @classmethod def get_currency(cls: Type[Currency], text: str) -> str | None: """ >>> Currency.get_currency(u'42') None >>> Currency.get_currency(u'42 €') u'EUR' >>> Currency.get_currency(u'$42') u'USD' >>> Currency.get_currency(u'42.000,00€') u'EUR' >>> Currency.get_currency(u'$42 USD') u'USD' >>> Currency.get_currency(u'%42 USD') u'USD' >>> Currency.get_currency(u'US1D') None """ curtexts = cls.EXTRACTOR.sub(' ', text.upper()).split() for currency, symbols in cls.CURRENCIES.items(): for curtext in curtexts: if curtext == currency: return currency for symbol in symbols: if curtext == symbol: return currency return None @classmethod def currency2txt(cls, currency): _currency = cls.CURRENCIES.get(currency, ('',)) return _currency[0]
[docs]def capability_to_string(capability_klass: Type[Capability]) -> str: m = re.match(r'^Cap(\w+)', capability_klass.__name__) assert m is not None return m.group(1).lower()
class DeprecatedFieldWarning(UserWarning): pass