Класс данных Python, представляющий собой смесь dict и namedtuple

Я хотел создать строго типизированный класс данных, который ведет себя как namedtuple и dict.

Он должен проверять типы данных переменных, используемых для его инициализации, и должен поддерживать индексацию в стиле последовательностей, поэтому он также должен быть итерируемым и поддерживать приведение типов к list. Он также должен поддерживать извлечение значений полей с использованием поля в качестве атрибута, в этом отношении он действует как именованный кортеж.

Однако он также должен поддерживать индексирование в стиле сопоставлений, используя имена полей в качестве ключей, поэтому он должен иметь возможность приведения типов к dict и сериализуемый JSON.

Я задал вопрос на StackOverflow, когда почти закончил свой код, но его закрыли из-за того, что он не сфокусирован. Сейчас я закончил его писать, но я действительно не знаю, плохая ли это идея. Я намерен использовать класс данных для хранения данных, извлеченных из базы данных SQLite3, где каждая строка соответствует экземпляру класса, чтобы сделать результаты запросов строго типизированными. И в базе данных буквально миллионы строк.

Я также хотел сделать его неизменным, но мне это удалось лишь частично.

Мой код:

import json
import time
from json import JSONEncoder
from typing import Any, Deque, Generator, Iterable, Mapping
script_start = time.time()


def json_default(self, obj):
    return getattr(obj.__class__, "__json__", json_default.default)(obj)
json_default.default = JSONEncoder().default

JSONEncoder.default = json_default

NoneType = type(None)
Network_Fields = (
    ('slash', str),
    ('start_integer', int),
    ('end_integer', int),
    ('start_string', str),
    ('end_string', str),
    ('count', int),
    ('ASN', (int, NoneType)),
    ('country_code', (str, NoneType)),
    ('is_anonymous_proxy', bool),
    ('is_anycast', bool),
    ('is_satellite_provider', bool),
    ('bad', bool)
)

Network_Main_Fields = (
    'slash', 'start_string',
    'end_string', 'ASN',
    'country_code', 'bad'
)

class DotDictTuple:
    def __init__(self, *args, **kwargs) -> None:
        self.__dict__['_values'] = []
        if args:
            assert not kwargs
            data = args if len(args) != 1 else args[0]
            self.from_dict(data) if isinstance(data, dict) else self.from_list(data)
        else:
            assert kwargs
            self.from_dict(kwargs)
    
    def from_list(self, sequence: Iterable) -> None:
        assert len(sequence) == self.__class__._field_count
        for item, (field, datatype) in zip(sequence, self.__class__._fields):
            assert isinstance(item, datatype)
            self.__dict__[field] = item
            self._values.append(item)
    
    def from_dict(self, mapping: Mapping) -> None:
        assert len(mapping) == self.__class__._field_count
        for field, datatype in self.__class__._fields:
            assert isinstance((item := mapping.get(field)), datatype)
            self.__dict__[field] = item
            self._values.append(item)
    
    def __getitem__(self, key: int | str | slice) -> Any:
        return self._values[key] if isinstance(key, (int, slice)) else self.__dict__[key]
    
    def __repr__(self) -> str:
        return f'{self.__class__.__name__}(' + ', '.join(
            f'{field}={self[field]!r}' for field in self.__class__._main_fields
        )+')'
    
    def __full_repr__(self) -> str:
        return f'{self.__class__.__name__}(' + ', '.join(
            f'{field}={self[field]!r}' for field in self.__class__._field_list
        )+')'
    
    def __len__(self) -> int:
        return self.__class__._field_count
    
    def __iter__(self) -> Generator:
        yield from self._values
    
    def __json__(self) -> dict:
        return {k: self.__dict__[k] for k in self.keys()}
    
    def values(self) -> tuple:
        return self._values
    
    def keys(self) -> tuple:
        return self.__class__._field_list
    
    def items(self) -> list:
        return [(k, self.__dict__[k]) for k in self.keys()]


class Network(DotDictTuple):
    _field_count = len(Network_Fields)
    _fields = Network_Fields
    _main_fields = Network_Main_Fields
    _field_list = [k for k, _ in Network_Fields]
    def __init__(self, *args, **kwargs) -> None:
        super(Network, self).__init__(*args, **kwargs)
        self.__dict__['_values'] = tuple(self._values)
    
    def __str__(self) -> str:
        return self.slash
    
    def __setattr__(self, *_, **__) -> None:
        raise TypeError
    
    __delattr__ = __setattr__

Пример использования

Network(['1.0.0.0/24', 16777216, 16777471, '1.0.0.0', '1.0.0.255', 256, 13335, 'AU', False, True, False, False])
Network(slash="1.0.0.0/24", start_integer=16777216, end_integer=16777471, start_string='1.0.0.0', end_string='1.0.0.255', count=256, ASN=13335, country_code="AU", is_anonymous_proxy=False, is_anycast=True, is_satellite_provider=False, bad=False)
Network(*['1.0.0.0/24', 16777216, 16777471, '1.0.0.0', '1.0.0.255', 256, 13335, 'AU', False, True, False, False])
Network('1.0.0.0/24', 16777216, 16777471, '1.0.0.0', '1.0.0.255', 256, 13335, 'AU', False, True, False, False)
Network({'slash': '1.0.0.0/24', 'start_integer': 16777216, 'end_integer': 16777471, 'start_string': '1.0.0.0', 'end_string': '1.0.0.255', 'count': 256, 'ASN': 13335, 'country_code': 'AU', 'is_anonymous_proxy': False, 'is_anycast': True, 'is_satellite_provider': False, 'bad': False})

network = Network(['1.0.0.0/24', 16777216, 16777471, '1.0.0.0', '1.0.0.255', 256, 13335, 'AU', False, True, False, False])
network.ASN
network['ASN']
network[0]
network['slash']
len(network)
list(network)
dict(network)
for i in network:
    print(i)

Все примеры работают.

Но я действительно не думаю, что мое решение является кратким и элегантным. Как сделать его более Pythonic?


Если я не пояснил, базовый класс предназначен для подкласса нескольких других классов, основная часть логики должна быть реализована в базовом классе, в то время как подклассы имеют разное количество полей и требования к типам.


Моя идея состоит в том, чтобы использовать базу данных SQLite3 только как способ сохранения данных между сеансами программы, другими словами, сериализовать данные.

У меня есть миллионы строк в данных, и json ужасно неэффективен для этой цели, и csv. Мне нужна библиотека бинарной сериализации, но pickle небезопасно и, что более важно, pkl, сгенерированный одной версией Python, не может использоваться другой версией.

Я не знаю, какая еще есть альтернатива. Я намереваюсь получить все строки из базы данных заранее, когда класс для запроса инициализирует его, поэтому все данные находятся в памяти, потому что я проверил, что извлечение одного значения из базы данных с использованием where занимает миллисекунды, мне нужно сделать это повторно.

2 ответа
2

Мне не нравится, что вы не можете различить поля/атрибуты class Network от взгляда на это. Вам нужно посмотреть на Network_Fieldsчто в данном случае далеко не где class Network определено.

Как предложил @Kache, может иметь смысл попытаться использовать dataclasses. Нам просто нужно добавить несколько методов, чтобы получить желаемое поведение.

класс данных с базовым классом

Один из способов сделать это — использовать базовый класс для добавления методов. Сложность в том, что класс не является «классом данных» до тех пор, пока @dataclass декоратор обрабатывает класс. Таким образом, любой базовый класс или метакласс не может использовать такие функции, как dataclasses.fields() чтобы найти все поля в классе данных. Это можно решить, проверив, полностью ли инициализирован класс при вызове метода экземпляра. Например:

class Base:
    def __getitem__(self, key):
        if isinstance(key, str):
            return getattr(self, key)  
        
        else:
            if not hasattr(self.__class__, '_name'):
                self.__class__._name = [f.name for f in fields(self)]

            if isinstance(key, slice):
                return [getattr(self, name) for name in self.__class__._name[key]]
            
            else:
                return getattr(self, self.__class__._name[key])    

            
    def __len__(self):
        return len(self.__class__._name)
        
        
@dataclass
class Bar(Base):
    field1: str
    field2: int
    field3: bool

При попытке доступа к полю Bar используя индекс или срез, класс проверяется на наличие _name атрибут. Если он не существует, создается один, содержащий список полей класса данных по порядку. Позволяет получить имена полей, соответствующие индексу или срезам. Это работает, но кажется немного «хакерским».

использовать декоратор

Другой вариант — использовать декоратор для добавления методов. Поскольку dataclass декоратор уже обработал класс, dataclasses.fields() Функция может использоваться для получения информации о полях, таких как имена, типы, значения по умолчанию и т. д.

# needed because None may be a valid value for a dataclass field
SENTINEL = object()

def makedictuple(cls):
    """decorator to add tuple-like and dict-like methods to a dataclass
    """
    
    # list mapping indexes to field names
    cls._name = [f.name for f in fields(cls)]

    def __getitem__(self, key):
        if isinstance(key, str):
            return getattr(self, key)  
        
        elif isinstance(key, slice):
            return [getattr(self, name) for name in cls._name[key]]
            
        else:
            return getattr(self, cls._name[key])    
        
    cls.__getitem__ = __getitem__
    
    
    def from_sequence(sequence):
        "a classmethod to construct cls from a sequence"
        
        for arg, field in zip(sequence, fields(cls)):
            if not isinstance(arg, field.type):
                raise TypeError(f"'{arg}' not of type '{field.type}'.")
        
        return cls(*sequence)
    
    cls.from_sequence = from_sequence

    def from_dict(mapping):
        "a classmethod to construct cls from a mapping"
        
        for field in fields(cls):
            value = mapping.get(field.name, SENTINEL)
            if value != SENTINEL and not isinstance(value, field.type):
                raise TypeError(f"Field ''{field.name}' value '{value}' not of type '{field.type}'.")
        
        return cls(**mapping)
        
    cls.from_dict = from_dict
    
    cls.__len__ = lambda self: len(cls._name)
    cls.__iter__ = lambda self: (getattr(self, name) for name in self._name)
    cls.keys = lambda self: asdict(self).keys()
    cls.values = lambda self: asdict(self).values()
    cls.items = lambda self: asdict(self).items()
    
    return cls

Используется следующим образом:

NoneType = type(None)

@makedictuple
@dataclass(frozen=True)
class Network:
    prefix: str
    start_integer: int
    end_integer: int
    start_string: str
    end_string: str
    count: int
    ASN: (int , NoneType)
    country_code: (str, NoneType)
    is_anonymous_proxy: bool
    is_anycast: bool
    is_satellite_provider: bool
    bad: bool

Все определяется в одном месте, и @dataclass декоратор создает __init__, __repr__, as_dict и другие методы.

    ('slash', str),

Хммм, интересное название для поля, похожего на "10.1.2.0/24". Подумайте о том, чтобы назвать его обычным именем prefix
(или, возможно, cidr, cidr_prefixили ip_prefix).

Попробуйте предложить для него один или два метода вывода с фиксированной шириной, чтобы мы получили «010.001.002.000/24» или шестнадцатеричное «0a010200/24». Это облегчило бы создание текста с разумным порядком сортировки.


    def __init__(self, *args, **kwargs) -> None:

гнида: mypy уже знает, что ctor возвращается None.


            assert isinstance((item := mapping.get(field)), datatype)

Как отмечает @RootTwo, оценивая assert для побочных эффектов на item
не имеет смысла. Предпочитать if ... raise ...


    def __full_repr__(self) -> str:

Пеп-8
просит, чтобы вы написали это _full_repr. Да, я получаю параллельную конструкцию для repr() метод. Но для этого нет соответствующего протокола, так что дандер вводит в заблуждение.


        return self.__class__._field_count

Может быть, в этом есть ценность документации? Но рассмотрите возможность использования более традиционного написания

        return self._field_count

Большая проблема в том, что DotDictTuple
не хватает нескольких атрибутов, включая _field_countпоэтому его, вероятно, следует отметить
абстрактный.


            self._values.append(item)
            ...
            self._values.append(item)
            ...
    def values(self) -> tuple:
        return self._values

Первые две строчки меня огорчают. (Также вызывая from_{list,dict} дважды приводит к плохому результату.)

Разве эта последняя строка не может получить необходимые значения, просматривая словарные статьи?


        self.__dict__['_values'] = tuple(self._values)

Я не понимаю эту строчку.

Если требуется кортеж, не должен ли родительский класс tuplified list когда все значения стали известны? Я возражаю против использования разных типов в родительских и дочерних классах.

Кроме того, это еще больше мотивирует получение значений из ключей dict.


Network(['1.0.0.0/24', 16777216, 16777471, '1.0.0.0', '1.0.0.255', 256, 13335, 'AU', False, True, False, False])
...
Network(*['1.0.0.0/24', 16777216, 16777471, '1.0.0.0', '1.0.0.255', 256, 13335, 'AU', False, True, False, False])

Я не понимаю, что желательно в поддержке обеих этих форм. Выберите один, последний. Удалить прежнее.

Network(slash="1.0.0.0/24", start_integer=16777216, end_integer=16777471, start_string='1.0.0.0', end_string='1.0.0.255', count=256, ASN=13335, country_code="AU", is_anonymous_proxy=False, is_anycast=True, is_satellite_provider=False, bad=False)
...
Network({'slash': '1.0.0.0/24', 'start_integer': 16777216, 'end_integer': 16777471, 'start_string': '1.0.0.0', 'end_string': '1.0.0.255', 'count': 256, 'ASN': 13335, 'country_code': 'AU', 'is_anonymous_proxy': False, 'is_anycast': True, 'is_satellite_provider': False, 'bad': False})

Точно так же замените 2-ю форму на Network(**{'slash': ...


    def __setattr__(self, *_, **__) -> None:
        raise TypeError

Слава! Мне нравится вся неизменная вещь.


Вы упомянули sqlite. Но я не заметил никаких
SqlAlchemy
интеграция. Следуйте совету @Kache. Я рекомендую вам написать несколько модульных тестов, которые просят SqlAlchemy хранить/извлекать строки в таблице sqlite. Обратите внимание на то, что говорит вам этот клиентский код. Если это неудобно или неуклюже писать, усовершенствуйте свой Public API, чтобы сделать его более удобным. Обратите внимание, что в течение for row in query: блок, достаточно часто просят row._asdict()или даже: for row in map(dict, query):


Этот код достигает своих целей проектирования.

Я был бы готов делегировать или принять задачи обслуживания этой кодовой базы.

Делиться

Улучшить этот ответ

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *