Make namedtuple accept kwargs

Question

If I have a class like:

class Person(object):
    def __init__(self, name, **kwargs):
        self.name = name

p = Person(name='joe', age=25) # age is ignored

Extra params are ignored. But if I have a namedtuple, I'll get `unexpected keyword argument:

from collections import namedtuple 

Person = namedtuple('Person', 'name')
p = Person(name='joe', age=25)

# Traceback (most recent call last):
#   File "python", line 1, in <module>
# TypeError: __new__() got an unexpected keyword argument 'age'

How can I make namedtuple accept kwargs so I can pass extra arguments safely?

Answer 1

It's not pretty:

p = Person(*(dict(name='joe', age=25)[k] for k in Person._fields))

Answer 2

It is possible to wrap the Person class constructor to ignore arguments which aren't defined as fields of the Person namedtuple:

from collections import namedtuple
Person = namedtuple('Person', 'name')

def make_person(*args, **kwargs):
    person_args = {}

    # process positional args
    if len(args) > len(Person._fields):
        msg = "Person() takes %d positional arguments but %d were given" % (len(Person._fields), len(args))
        raise TypeError(msg)
    for arg_name, arg_value in zip(Person._fields, args):
        person_args[arg_name] = arg_value

    # process keyword args
    for arg_name, arg_value in kwargs.items():
        try:
            i = Person._fields.index(arg_name)
        except ValueError:
            pass # ignore arguments not defined as Person fields
        else:
            if arg_name in person_args:
                msg = "make_person() got multiple values for argument " + repr(arg_name)
                raise TypeError(msg)
            person_args[arg_name] = arg_value

    if len(person_args) != len(Person._fields):
        msg = "Person() requires additional arguments: "
        msg += ", ".join([repr(x) for x in Person._fields if x not in person_args])
        raise TypeError(msg)
    return Person(*[person_args[x] for x in Person._fields])

Given the above:

>>> make_person('a')
Person(name='a')
>>> make_person('a', b='b')
Person(name='a')
>>> make_person('a', name='b')
TypeError: make_person() got multiple values for argument 'name'
>>> make_person(b='b')
TypeError: Person() requires additional arguments: 'name'
>>> make_person(1, 2)
TypeError: Person() takes 1 positional arguments but 2 were given

Answer 3

A slight variation of @paulmcg answer using a factory method:

_Person = namedtuple('Person', 'name')

class Person(_Person):
    @staticmethod
    def from_dict(args):
        args = {k: v for k, v in args.items() if k in _Person._fields}
        return Person(**args)


p = Person.from_dict(dict(name='joe', age=25))
print(p)

Answer 4

Straightforward, just give your namedtuple a kwargs field and set its default value to an empty dictionary. Then any key-value pair can be added and accessed under that kwargs field, see the sample code below.

from collections import namedtuple 

Person = namedtuple('Person', 'name, kwargs', defaults=['', {}])
p = Person(name='joe', kwargs={'age': 25})

print(p)
print(p.kwargs['age'])

p2 = Person()

print(p2)

Output:

Person(name='joe', kwargs={'age': 25})

25

Person(name='', kwargs={})

Answer 5

It appears these settings will use multiple partitions: this will consume all messages.

mp.messaging.incoming.your-events.auto.offset.reset=earliest
mp.messaging.incoming.your-events.group.id=${quarkus.uuid}

If you are using an emitter this will work without the above settings;

int partition = 0;
Message<Integer> message = Message.of(value)
            .addMetadata(OutgoingKafkaRecordMetadata.<String>builder()
                .withKey(key)
                .withPartition(partition) // change for each partition, 0, 1, 2..                
                .withTopic("your-events")
                .build());