Convenience facilities for objects.

Short summary:

• as_dict: Return a dictionary with keys mapping to the values of the attributes of o.

• copy: Convenient function to shallow copy an object with simple modifications.

• flavour: Return constants indicating the ``flavour'' of an object: * T_MAP: DictType, DictionaryType, objects with an keys or keys attribute. * `T_SEQ`: TupleType, ListType, objects with an iter attribute. * `T_SCALAR`: Anything else.

• NoAttrs: A stub class whose __getattr__ raises AttributeError. This exists to be put at the end of the subclasses for concrete nonmixin classes which have multiple __getattr__ methods in their MRO, so that they can always call super().__getattr__.

• O: The O class is now obsolete, please subclass types.SimpleNamespace or use a dataclass.

• O_attritems: Generator yielding (attr,value) for relevant attributes of o.

• O_attrs: Yield attribute names from o which are pertinent to O_str.

• O_merge: Merge key:value pairs from a mapping into an object.

• O_str: Return a str representation of the object o.

• obj_as_dict: OBSOLETE version of obj_as_dict, suggestion: use cs.obj.as_dict.

• Proxy: An extremely simple proxy object that passes all unmatched attribute accesses to the proxied object.

• public_subclasses: Return a set of the subclasses of cls which have public names.

• Refreshable: A mixin for refreshable objects. The object must provide a _refresh() method, returning True if the object was updated and false otherwise.

• Sentinel: A simple class for named sentinels whose str() is just the name and whose == uses is.

• singleton: Obtain an object for key via registry (a mapping of key=>object). Return (is_new,object).

• SingletonMixin: A mixin turning a subclass into a singleton factory.

• TrackedClassMixin: A mixin to track all instances of a particular class.

Functions

O_attritems(o)

Generator yielding (attr,value) for relevant attributes of o.

O_attrs(o)

Yield attribute names from o which are pertinent to O_str.

Note: this calls getattr(o,attr) to inspect it in order to prune callables.

O_merge(o, _conflict=None, _overwrite=False, **kw)

Merge key:value pairs from a mapping into an object.

Ignore keys that do not start with a letter. New attributes or attributes whose values compare equal are merged in. Unequal values are passed to:

_conflict(o, attr, old_value, new_value)

to resolve the conflict. If _conflict is omitted or None then the new value overwrites the old if _overwrite is true.

O_str(o, no_recurse=False, seen=None)

Return a str representation of the object o.

Parameters:

• o: the object to describe.
• no_recurse: if true, do not recurse into the object's structure. Default: False.
• seen: a set of previously sighted objects to prevent recursion loops.

as_dict(o, selector=None)

Return a dictionary with keys mapping to the values of the attributes of o.

Parameters:

• o: the object to map
• selector: the optional selection criterion

If selector is omitted or None, select "public" attributes, those not commencing with an underscore.

If selector is a str, select attributes starting with selector.

Otherwise presume selector is callable and select attributes attr where selector(attr) is true.

copy(obj, **kw)

Convenient function to shallow copy an object with simple modifications.

Performs a shallow copy of self using copy.copy.

Treat all keyword arguments as (attribute,value) 2-tuples and replace those attributes with the supplied values.

flavour(obj)

Return constants indicating the ``flavour'' of an object:

• T_MAP: DictType, DictionaryType, objects with an keys or keys attribute.
• T_SEQ: TupleType, ListType, objects with an iter attribute.
• T_SCALAR: Anything else.

obj_as_dict(o, **kw)

OBSOLETE version of obj_as_dict, suggestion: use cs.obj.as_dict

OBSOLETE convesion of an object to a dict. Please us cs.obj.as_dict.

public_subclasses(cls, extras=())

Return a set of the subclasses of cls which have public names.

singleton(registry, key, factory, fargs, fkwargs)

Obtain an object for key via registry (a mapping of key=>object). Return (is_new,object).

If the key exists in the registry, return the associated object. Otherwise create a new object by calling factory(*fargs,**fkwargs) and store it as key in the registry.

The registry may be any mapping of keys to objects but will usually be a weakref.WeakValueDictionary in order that object references expire as normal, allowing garbage collection.

Note: this function is not thread safe. Multithreaded users should hold a mutex.

See the SingletonMixin class for a simple mixin to create singleton classes, which does provide thread safe operations.

Classes

class NoAttrs

A stub class whose __getattr__ raises AttributeError. This exists to be put at the end of the subclasses for concrete nonmixin classes which have multiple __getattr__ methods in their MRO, so that they can always call super().__getattr__.

This exists because object has no __getattr__, necessitating this ugly shuffle in mixin __getattr__ methods:

def __getattr__(self, attr):
    ... return the mixin's special attr ...
    try:
        gsa = super().__getattr__
    except AttributeError:
        raise AttributeError(f'{self.__class__.__name__}.{attr}')
    else:
        return gsa(attr)

With NoAttrs at the end of the MRO every mixin's __geattr__ can be:

def __getattr__(self, attr):
    ... return the mixin's special attr ...
    return super().__getattr__(attr)

A concrete class is then defined as:

class C(AMixin,BMixin,NoAttrs):

Because of Python's MRO construction rules, a subclass like:

class D(C,DMixin,NoAttrs):

still places NoAttrs at the end of the MRO, producing correct behaviour.

NoAttrs.__getattr__ issues a warning if super().__getattr__ actually exists.

NoAttrs.__getattr__(self, attr)

Just raise AttributeError with a nice mesage.

class O(types.SimpleNamespace)

The O class is now obsolete, please subclass types.SimpleNamespace or use a dataclass.

O.callers

set()

class Proxy

An extremely simple proxy object that passes all unmatched attribute accesses to the proxied object.

Note that setattr and delattr work directly on the proxy, not the proxied object.

class Refreshable

A mixin for refreshable objects. The object must provide a _refresh() method, returning True if the object was updated and false otherwise.

This mixin provides a .refresh() method which implements the refresh logic and various related refresh_* methods/properties.

Refreshable.REFRESH_LIFESPAN

604_800

Refreshable.REFRESH_RATELIMIT

1.0

Refreshable.refresh(self, *, data=None, force=False, lifespan: float | None = None, map=None, concurrent=False, ratelimit: float | None = None, recurse=False, seen=None, **_refresh_kw) -> bool

Refresh this object; if it is stale attempt to refresh via self._refresh(). Return True if the object was updated, False otherwise.

Note that this returns False if the object was considered still current; False is not an error.

The refresh policy is: if force or (the object's information is stale and the rate limit does not preclude a refresh), call self._refresh(). This is measured by self.refresh_needed().

State about the refresh poll times is kept in self.refresh_last_poll and self.refresh_last_update.

Keyword parameters:

• data: optional object providing the refresh data, passed to self._refresh() as the data keyword argument; if not None then _refresh is always called and should apply it instead of making whatever API call or other action it would perform passed to `self._refresh()
• force: optional flag, default False; if true attempt a refresh regardless of staleness or the rate limit
• lifespan: how many seconds before updated information is considered no stale, default from self.refresh_lifespan, or type(self).REFRESH_RATELIMIT
• map: optional map()-like function to call refresh on the related objects; this hook exists to allow for example APIs to pass in a curried pmap() call to refresh some objects in parallel; the default map function is map()
• ratelimit: how many seconds should elapsed before attempting a refresh, default from self.refresh_ratelimit or type(self).REFRESH_RATELIMIT
• recurse: optional flag, default False; if true then refresh the objects from self.refresh_related1() nonrecursively and recursively refresh the objects from self.refresh_related()
• seen: optional set of keys to prevent unbound recursion

Other keyword parameters are passed to self._refresh() if it is called.

Note the two related-object methods, refresh_related() and refresh_related1(). These exist to allow a recursive refresh to refresh "descendant" objects recursively and other related "adjacent" or "superior" objects nonrecursively.

Supposing you're refreshing a section topic in a news site; you might recursively refresh all the articles in the topic and the authors of the articles. However you likely also want to refresh other objects related to the topic, perhaps the table of topics and other things; recursing into those objects might refresh everything on the entire site. Accordingly the descendant objects would be enumerated by refresh_related() and the adjacent/superior objects by refresh_related1().

Note that because a recurse=True call descends a potentially arbitrary tree of related objects, if map is supplied as, for example, pmap, that should be a curried call with a shared Semaphore to control the concurrency, such as partial(pmap,concurrent=Semaphore(4)).

It is probably a mistake to call this with non-None data and recurse true. It will be accepts and the data applied to self and the recursion done without data. (Now we raise ValueError for this.)

Refreshable.refresh_bulk(objs: Iterable[ForwardRef('Refreshable')], *, concurrent=False, map=None, **refresh_kw)

Refresh multiple Refreshable objects, concurrently if concurrent is not False.

Refreshable.refresh_key(self)

Return the key value for the seen set used in a recursive refresh. This default returns id(self); classes like Entity would use a characteristic value like self.name.

Refreshable.refresh_needed(self, *, lifespan: float | None = None, now: float | None = None) -> bool

Test whether self is considered stale.

Parameters:

• lifespan: how many seconds before updated information is considered no stale, default from self.refresh_lifespan, or type(self).REFRESH_LIFESPAN
• now: optional reference time, default from time.time()

Refreshable.refresh_related(self) -> Iterable

Return the related objects which should also be refreshed in recursive refreshes.

Refreshable.refresh_related1(self) -> Iterable

Return the related objects which should also be refreshed in recursive refreshes, but nonrecursively.

Refreshable.refreshed(self, **refresh_kw)

A variant on refresh() which returns self to support chaining. This isn't the default because that's not Pythonic.

class Sentinel

A simple class for named sentinels whose str() is just the name and whose == uses is.

Example:

>>> from cs.obj import Sentinel
>>> MISSING = Sentinel("MISSING")
>>> print(MISSING)
MISSING
>>> other = Sentinel("other")
>>> MISSING == other
False
>>> MISSING == MISSING
True

Sentinel.__hash__

None

class SingletonMixin

A mixin turning a subclass into a singleton factory.

Note: this mixin overrides object.__new__ and may not play well with other classes which override __new__.

Warning: because of the mechanics of __new__, the instance's __init__ method will always be called after __new__, even when a preexisting object is returned. Therefore that method should be sensible even for an already initialised and probably subsequently modified object.

My suggested approach is to access some attribute, and preemptively return if it already exists. Example:

def __init__(self, x, y):
    if 'x' in self.__dict__:
        return
    self.x = x
    self.y = y

Note: we probe self.__dict__ above to accomodate classes with a __getattr__ method.

Note: each class registry has a lock, which ensures that reuse of an object in multiple threads will call the __init__ method in a thread safe serialised fashion.

Implementation requirements: a subclass should:

• provide a method _singleton_key(*args,**kwargs) returning a key for use in the single registry, computed from the positional and keyword arguments supplied on instance creation i.e. those which __init__ would normally receive. This should have the same signature as __init__ but using cls instead of self.
• provide a normal __init__ method which can be safely called again after some earlier initialisation.

This class is thread safe for the registry operations.

Example:

class Pool(SingletonMixin):

    @classmethod
    def _singleton_key(cls, foo, bah=3):
        return foo, bah

    def __init__(self, foo, bah=3):
        if hasattr(self, 'foo'):
            return
       ... normal __init__ stuff here ...
       self.foo = foo
       ...

SingletonMixin.__hash__(self)

default hash and equality methods

SingletonMixin.singleton_also_by(also_key, key)

Obtain a singleton by a secondary key. Return the instance or None.

Parameters:

• also_key: the name of the secondary key index
• key: the key for the index

class TrackedClassMixin

A mixin to track all instances of a particular class.

This is aimed at checking the global state of objects of a particular type, particularly states like counters. The tracking is attached to the class itself.

The class to be tracked includes this mixin as a superclass and calls:

TrackedClassMixin.__init__(class_to_track)

from its init method. Note that class_to_track is typically the class name itself, not type(self) which would track the specific subclass. At some relevant point one can call:

self.tcm_dump(class_to_track[, file])

class_to_track needs a tcm_get_state method to return the salient information, such as this from cs.resources.MultiOpenMixin:

def tcm_get_state(self):
    return {'opened': self.opened, 'opens': self._opens}

See cs.resources.MultiOpenMixin for example use.

TrackedClassMixin.tcm_all_state(klass)

Generator yielding tracking information for objects of type klass in the form (o,state) where o if a tracked object and state is the object's get_tcm_state method result.

TrackedClassMixin.tcm_dump(klass, f=None)

Dump the tracking information for klass to the file f (default sys.stderr).