2
\$\begingroup\$

I wanted to create a data class that is strongly typed, and behaves both like a namedtuple and dict.

It should check the data types of the variables used to initialize it, and should support indexing in the style of sequences, therefore it should also be iterable and support type casting to list. It should also support retrieving values of the fields by using the field as attributes, in this regard it acts like a namedtuple.

However it should also support indexing in the style of mappings, by using the field names as keys, so it should be able to type cast to dict and JSON serializable.

I asked a question on StackOverflow when I almost finished my code, but it got closed for being unfocused. Now I have finished writing it, but I really don't know if this is a bad idea. I intend to use the data class to hold data retrieved from an SQLite3 database, where each row corresponds to an instance of the class, to make the results of queries strongly typed. And there are literally millions of rows in the database.

I also wanted to make it immutable, but I only partly succeeded doing so.

My code:

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__

Example usage

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)

All the examples work.

But I really don't think my solution is concise and elegant. How to make it more Pythonic?


If I didn't make it clear, the base class is intended to be subclassed by a few other classes, the bulk of the logic should be implemented in the base class, while the subclasses have different numbers of fields and type requirements.


My idea is to only use the SQLite3 database as a way to persist the data between sessions of the program, in other words serializing the data.

I have millions of rows in the data, and json is terribly inefficient for this purpose, and is csv. I need a binary serialization library, but pickle is unsafe and more importantly a pkl generated by one Python version cannot be used by another version.

I don't know what other alternative there is. I intend to retrieve all rows from the database upfront when the class to query it initializes, so all data is in memory, because I have tested that retrieving a single value from the database using where takes milliseconds, I need to do it repeatedly.

\$\endgroup\$
2
  • \$\begingroup\$ I think it'd be better if you built on top of standard dataclasses and typing rather than hand-rolled your own. You also mention these represent database rows, and sqlalchemy also already has integrations with dataclasses. \$\endgroup\$
    – Kache
    Commented May 11, 2023 at 21:26
  • \$\begingroup\$ This is a bad idea: assert isinstance((item := mapping.get(field)), datatype). When someone runs the code with the optimization flags, -O or -OO, assert statements are removed, so item won't be assigned, which could lead to various exceptions or worse, wrong data from item defined in an outer scope. \$\endgroup\$
    – RootTwo
    Commented May 12, 2023 at 0:01

2 Answers 2

1
\$\begingroup\$

I dislike that you can't discern the fields/attributes of class Network from looking at it. You need to look at Network_Fields, which in this case is not near where class Network is defined.

As @Kache suggested, it might make sense to try to use dataclasses. We just need to add some methods to get the desired behaviors.

dataclass with a base class

One way to do that us to use a base class to add the methods. The difficulty is that the class isn't a "dataclass" until after the @dataclass decorator processes the class. So any base class or meta class can't use functions like dataclasses.fields() to find all the fields in the dataclass. This can be handled by testing if the class is fully initialized when an instance method is called. For example:

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

When an attempt is made to access a field of Bar using an index or a slice, the class is checked for a _name attribute. If it doesn't exist one is created containing a list of the dataclass fields in order. The enables one to get the field names corresponding to the index or slices. It works but seems a little "hacky".

use a decorator

Another option is to use a decorator to add the methods. Because, the dataclass decorator has already processed the class, the dataclasses.fields() function can be used to get information about the fields, such as names, types, defaults, etc.

# 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

Used like this:

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

Everything is defined in one place, and the @dataclass decorator creates __init__, __repr__, as_dict and other methods.

\$\endgroup\$
1
\$\begingroup\$
    ('slash', str),

Hmmm, interesting name for a field that looks like "10.1.2.0/24". Consider calling it by its conventional name of prefix (or perhaps cidr, cidr_prefix, or ip_prefix).

Consider offering one or two fixed-width output methods for it, so we get "010.001.002.000/24", or hex "0a010200/24". That would facilitate producing text which has a sensible sort order.


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

nit: mypy already knows that ctor returns None.


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

As @RootTwo observes, evaluating an assert for side effects on item doesn't make sense. Prefer if ... raise ...


    def __full_repr__(self) -> str:

Pep-8 asks that you spell it _full_repr. Yes, I get the parallel construction for the repr() method. But there's no corresponding protocol to invoke this, so the dunder is misleading.


        return self.__class__._field_count

Maybe there's documentation value in this? But consider using the more conventional spelling of

        return self._field_count

The larger issue is that DotDictTuple lacks several attributes including _field_count, so it should probably be marked abstract.


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

Those first two lines make me sad. (Also, invoking from_{list,dict} twice leads to a bad result.)

Couldn't that final line derive the needed values by walking through dictionary entries?


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

I don't understand this line.

If a tuple is desired, shouldn't the parent class have tuplified the list when all values became known? I'm objecting to using different type in the parent & child classes.

Also, this further motivates deriving values from dict keys.


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])

I don't understand what's desirable about supporting both those forms. Pick one, the latter. Delete the former.

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})

Similarly, replace the 2nd form with Network(**{'slash': ...


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

Kudos! I like the whole immutable thing.


You mentioned sqlite. But I didn't notice any SqlAlchemy integration. Do follow @Kache's advice. I recommend you code up some unit tests that ask SqlAlchemy to store / retrieve rows in a sqlite table. Pay attention to what that client code is telling you. If it's inconvenient or clunky to write, then refine your Public API to make it more convenient. Note that within a for row in query: block, it is common enough to ask for row._asdict(), or even: for row in map(dict, query):


This code achieves its design goals.

I would be willing to delegate or accept maintenance tasks on this codebase.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.