Source code for aiomixcloud.models

"""
Basic data structures
~~~~~~~~~~~~~~~~~~~~~

This module contains the basic data structures used throughout
the package.  Specifically:

    - :class:`_WrapMixin`, a mixin providing type casting of
      accessed items.

    - :class:`AccessDict`, dict-like, supports accessing items
      by attribute.  Accessed items are properly wrapped.

    - :class:`AccessList`, list-like, supports accessing
      :class:`Resource` items by their "key" key.  Accessed items
      are properly wrapped.

    - :class:`Resource`, like an :class:`AccessDict` which has
      a "type" key.  Gets methods for downloading its "connections",
      that is the sub-entities "owned" by the resource.
      Mirrors methods of :class:`~aiomixcloud.core.Mixcloud` marked as
      "targeted" with their first argument as its key.  Also provides
      a :meth:`Resource.load` method to download full resource
      information in case it is partly loaded as an element of another
      API response.

    - :class:`ResourceList`, like an :class:`AccessDict` which
      delegates string indexing and iterating over, to its "data" item.
      Supports pagination through the :meth:`ResourceList.previous`
      and :meth:`ResourceList.next` methods, which return a new object
      with the respective data.
"""

from collections import UserDict, UserList
from functools import partial
from types import MethodType

from aiomixcloud.decorators import paginated


class _WrapMixin:
    """Enables returning the proper kind of object, when indexed and
    iterated over.  Produces aiomixcloud models out of dictionaries
    and lists, leaving the rest of the types intact.
    """

    def __init__(self, data, *, mixcloud):
        """Call super's `__init__` and store
        :class:`~aiomixcloud.core.Mixcloud` instance, if given one.
        """
        super().__init__(data)
        #: :class:`~aiomixcloud.models.Mixcloud` instance to pass
        #: along to contained items
        self.mixcloud = mixcloud

    def __getitem__(self, key):
        """Wrap and return item gotten by `key`."""
        item = super().__getitem__(key)
        return self._wrap(item)

    def __iter__(self):
        """Wrap yielded values."""
        for item in super().__iter__():
            yield self._wrap(item)

    def _wrap(self, item):
        """Wrap `item` with proper class.  `dict` becomes
        :class:`AccessDict`, or :class:`Resource`-like if it has
        ``"type"`` as a key.  `list` becomes :class:`AccessList`.
        The rest remain unchanged.
        """
        if isinstance(item, dict):
            if 'type' in item:
                # Item is considered an entity, make it a resource.
                return self.mixcloud._resource_class(
                    item, mixcloud=self.mixcloud)
            return AccessDict(item, mixcloud=self.mixcloud)

        if isinstance(item, list):
            return AccessList(item, mixcloud=self.mixcloud)

        return item


[docs]class AccessDict(_WrapMixin, UserDict): """Dict-like model which supports accessing items using keys as attributes. Items are wrapped with a proper model, depending on their type. Original `dict` is stored in `self.data`. """ def __getattr__(self, name): """Try returning an item with given `name` as a key.""" try: return self.__getitem__(name) except KeyError: raise AttributeError(f'{self.__class__.__name__!r} object ' f'has no attribute {name!r}') from None
[docs]class AccessList(_WrapMixin, UserList): """List-like model which supports accessing :class:`Resource`-like items by matching their "key" item. Items are wrapped with a proper model, depending on their type. Original `list` is stored in `self.data`. """ def __getitem__(self, key): """Try returning item by given `key`. On failure, try to find and return a :class:`Resource`-like item whose their "key" item equals `key`. Raise :exc:`KeyError` on failure of finding one. """ try: return super().__getitem__(key) except TypeError: # `key` is probably a string, try to find a resource # whose key matches it. # Surround `key` by slashes in case it is not already # surrounded. if not key.startswith('/'): key = f'/{key}' if not key.endswith('/'): key = f'{key}/' for item in self: if (isinstance(item, self.mixcloud._resource_class) and item['key'] == key): return item raise KeyError(key) from None
[docs]class Resource(AccessDict): """Mixcloud API resource A resource is like an :class:`AccessDict` object which has a "type" key. When a "type" key is present in an API (sub)object, suggesting it has a unique URL, it is considered an API resource, that is an individual entity (a user, a cloudcast, a tag, etc). A :class:`Resource` object has appropriately named methods for downloading information about its sub-entities ("connections"). It also mirrors "targeted" methods of its :class:`~aiomixcloud.core.Mixcloud` instance, passing them its key as a first argument. Targeted methods include "actions" (e.g :meth:`~aiomixcloud.core.Mixcloud.follow` or :meth:`~aiomixcloud.core.Mixcloud.unfavorite`), embed-related methods and :meth:`~aiomixcloud.core.Mixcloud.edit`. """ def __init__(self, data, *, full=False, create_connections=True, mixcloud): """Pass `mixcloud` to super's `__init__` and store whether resource is full. If it is full and `create_connections` is set, create resource connections. """ super().__init__(data, mixcloud=mixcloud) #: Whether all of resource data has been downloaded (by having #: accessed the detail page). self._full = full if full and create_connections: self._create_connections() def __getattr__(self, name): """If super fails to find an attribute named `name`, try to find a method of :attr:`mixcloud` with the `_targeted` attribute set and return a version of it with the first argument frozen as current object's key. """ try: return super().__getattr__(name) except AttributeError: mixcloud_attribute = getattr(self.mixcloud, name, None) if hasattr(mixcloud_attribute, '_targeting'): # Targeting method of Mixcloud instance found, # freeze its first argument as own key and return it. return partial(mixcloud_attribute, self['key']) # Targeting method not found, let AttributeError # pass through. raise def __repr__(self): """Return representation string consisting of class name, resource type and value of the "key" key. """ # Make resource type friendlier to read resource_type = self['type'].replace('_', ' ').title() return f'<{self.__class__.__name__}: {resource_type} {self["key"]!r}>' def _create_connections(self): """In case there is an item with a ``['metadata']['connections']`` key, create a method for each of its items (resource "connections") that fetches information about sub-entities associated with the resource (eg `comments`, `followers` etc). Each of these methods is named after the respective connection. """ try: connections = self['metadata']['connections'] except KeyError: pass else: # Make a method for each connection of the object. # Use a factory function to avoid late binding. def make_fetcher(url): """Return a function that fetches information from `url`. """ @paginated async def fetcher(self, params): """Download sub-entities information from `url` and return the relevant :class:`ResourceList`. """ return await self.mixcloud.get( url, relative=False, **params) return fetcher for name, url in connections.items(): fetcher_function = make_fetcher(url) # Make produced function a method and bind it to `self`. method = MethodType(fetcher_function, self) # Make sure it has a valid identifier. name = name.replace('-', '_') setattr(self, name, method)
[docs] async def load(self, *, force=False): """Load full resource information from detail page. Do nothing in case :attr:`_full` is ``True``, unless `force` is set. Return `self`, so this can be used in chained calls. """ if not self._full or force: full_resource = await self.mixcloud.get( self['key'], create_connections=False) self.update(full_resource) self._create_connections() self._full = True return self
[docs]class ResourceList(AccessDict): """Contains a list of resources, with paging capabilities. Main data is stored in the ``'data'`` item, while a ``'paging'`` item may be present indicating URLs of the previous and next pages of the resource list, as well as a `'name'` item describing the collection. Indexing falls back to ``self['data']`` on failure, while iterating over and length concern "data" straight up. """ def __getitem__(self, key): """If `key` is not found in `self`, delegate to ``self['data']`` (list of contained resources). """ try: return super().__getitem__(key) except KeyError: return self['data'].__getitem__(key) def __iter__(self): """Iterate over contained resources.""" return self['data'].__iter__() def __len__(self): """Return count of contained resources.""" return self['data'].__len__() def __repr__(self): """Return representation string consisting of class name and value of the "name" key, if it exists. """ display = self.__class__.__name__ if 'name' in self: display = f'{display} {self["name"]!r}' return f'<{display}>' async def _navigate(self, where): """Return an adjacent page of current resource list (another :class:`ResourceList` object) specified by `where`, or ``None`` if it is not found. """ try: url = self['paging'][where] except KeyError: return None return await self.mixcloud.get(url, relative=False)
[docs] async def previous(self): """Return previous page of current resource list, or ``None`` if it is not found. """ return await self._navigate('previous')
[docs] async def next(self): """Return next page of current resource list, or ``None`` if it is not found. """ return await self._navigate('next')