from __future__ import annotations
from typing import TYPE_CHECKING, Generic, Type, TypeVar
from uuid import UUID
import pandas as pd
from .proxy import Proxy
if TYPE_CHECKING:
from ..client import Client
from .refs import RefList
ProxyClass = TypeVar("ProxyClass", bound=Proxy)
[docs]
class ShortenedUUID(UUID):
def __str__(self):
return f"{super().__str__()[:5]}..."
[docs]
def simplified(value, property):
from .property import Id
from .refs import Reference
match property:
case Id():
return ShortenedUUID(value.hex)
case Reference(proxy_type=pt):
if pt.proxy_schema.name_id:
return value.name
elif isinstance(value, Proxy):
return ShortenedUUID(value.proxy_id.hex)
elif isinstance(value, ProxyList):
return f"{pt.proxy_schema.class_name}[{len(value)}]"
else:
return value
case _:
return value
[docs]
class ProxyList(Generic[ProxyClass]):
"""
Base class for "dynamic lists" of proxies.
The instances maintain lists of IDs and translate list operations
as if the list actually contained proxy objects.
"""
def __init__(self, client: Client, proxy_type: Type[ProxyClass]):
self.client = client
self.proxy_type = proxy_type
self.registry = client.registry_for(proxy_type)
@property
def coll(self) -> list[UUID]:
raise NotImplementedError
[docs]
def resolve_proxy(self, item):
"""This is the main routine that transforms elements of
the list to proxies.
"""
return self.registry.fetch_proxy(item)
[docs]
def get_slice(self, slc: slice) -> list[UUID]:
"""Return a list of UUIDs for the given slice.
This method is used to implement slicing operations on the
ProxyList. It should be overriden in subclasses depending on
the underlying implementation the the collection.
"""
raise ValueError("Slices are not supported yet")
def __iter__(self):
for item in self.coll:
yield self.resolve_proxy(item)
def __len__(self):
return len(self.coll)
def __getitem__(self, item):
if isinstance(item, slice):
return ProxyVec(self.client, self.proxy_type, self.get_slice(item))
else:
return self.resolve_proxy(self.coll[item])
def __repr__(self):
return f"{self.proxy_type.__name__}{repr(self.coll)}"
def __eq__(self, other):
try:
return len(self) == len(other) and (
all(p == q for p, q in zip(self, other))
)
except Exception:
return False
@property
def ids(self):
return list(self.coll)
[docs]
def to_df(self, *additional_fields, fields=None, simplify=True):
"""Generate a pandas dataframe for the list of proxy entities.
The dataframe is generated by fetching and tabulating a
subset of fields, for each entity in the list.
"""
schema = self.proxy_type.proxy_schema
if fields is None:
fields = schema.short_list(set(additional_fields))
elif fields in (True, "all", "ALL"):
fields = list(schema.all_fields) + list(additional_fields)
else:
fields = list(fields) + list(additional_fields)
data = {field: list() for field in fields}
for proxy in self:
for field in fields:
if simplify:
property = schema.all_fields[field]
data[field].append(
simplified(getattr(proxy, field, pd.NA), property)
)
else:
data[field].append(getattr(proxy, field, pd.NA))
return pd.DataFrame(data=data)
@property
def df(self):
"""Return a dataframe over the default fields."""
return self.to_df()
@property
def DF(self):
"""Return a dataframe over all fields."""
return self.to_df(fields=True, simplify=False)
[docs]
class ProxyVec(ProxyList):
"""A list of IDs appearing as proxies.
The underlying data is a list of UUIDs. At each element access, the
correpsonding element is fetched from the registry.
"""
def __init__(
self, client: Client, proxy_type: Type[ProxyClass], members: list[UUID]
):
"""Initialize the vector with the client, the proxy type, and the list of IDs.
Arguments:
client: the client object
proxy_type: the type of the proxies
members: the list of UUIDs
"""
super().__init__(client, proxy_type)
self.members: list[UUID] = members
@property
def coll(self):
return self.members
[docs]
def get_slice(self, slc: slice) -> list[UUID]:
return self.members[slc]
[docs]
class ProxySublist(ProxyList):
"""A proxy class that translates collection operations to operations
on an entity sub-collection.
"""
def __init__(self, property: RefList, owner: Proxy):
super().__init__(owner.proxy_registry.catalog, property.proxy_type)
self.property = property
self.owner = owner
def __delitem__(self, key):
raise NotImplementedError(
f"delitem {self.property.owner.__name__}.{self.property.name}"
)
def __iadd__(self, **kwargs):
raise NotImplementedError(
f"iadd {self.property.owner.__name__}.{self.property.name}"
)
@property
def coll(self):
return self.property.get(self.owner)
[docs]
def get_slice(self, slc: slice) -> list[UUID]:
return self.property.get(self.owner)[slc]