Module `piate.api.resources.inventories`

Expand source code

from dataclasses import dataclass, field
from typing import Optional, TypeVar, Generator, List, Callable, TypedDict

from dataclasses_json import dataclass_json, Undefined

from piate.api import API
from piate.api.resources import BaseResource
from piate.api.response import Meta

T = TypeVar("T")


@dataclass_json(undefined=Undefined.RAISE)
@dataclass
class InventoryElement:
    meta: Meta
    code: str
    name: str
    is_official: Optional[bool] = field(default=None)
    deprecated: Optional[bool] = field(default=None)

    class Compact(TypedDict):
        code: str
        name: str
        deprecated: Optional[bool]
        is_official: Optional[bool]

    def compact(self) -> Compact:
        compact = {"code": self.code, "name": self.name}
        if self.is_official is not None:
            compact["is_official"] = self.is_official
        if self.deprecated:
            compact["deprecated"] = self.deprecated
        return compact


@dataclass_json(undefined=Undefined.RAISE)
@dataclass
class InventoryElementsResponse:
    items: List[InventoryElement]

    meta: Meta

    offset: Optional[int] = field(default=None)
    limit: Optional[int] = field(default=None)
    size: Optional[int] = field(default=None)

    next: Optional[Meta] = field(default=None)

    def compact(self) -> List[InventoryElement.Compact]:
        return [item.compact() for item in self.items]


def pages_inventory_method(
    path: str,
) -> Callable[
    ["Inventories", Optional[str]], Generator[InventoryElementsResponse, None, None]
]:
    return lambda self, translation_language=None: self._pages(
        f"/inventories/_{path}", translation_language
    )


def get_inventory_method(
    path: str,
) -> Callable[
    ["Inventories", Optional[str], Optional[str], Optional[bool]], InventoryElement
]:
    def inventory_method(
        self: "Inventories",
        code: Optional[str],
        name: Optional[str],
        is_official: Optional[bool],
    ) -> InventoryElement:

        official_func = lambda l: (is_official is None) or (
            l.is_official == is_official
        )

        filter_by_name = lambda l: (l.name == name) and official_func(l)
        filter_by_code = lambda l: (l.code == code) and official_func(l)
        filter_func = filter_by_name if code is None else filter_by_code

        response = {}
        for page in self._pages(f"/inventories/_{path}"):
            for language in page.items:
                if filter_func(language):
                    response = language
        return response

    return inventory_method


class Inventories(BaseResource):
    get_language = get_inventory_method("languages")
    pages_languages = pages_inventory_method("languages")

    get_term_type = get_inventory_method("term-types")
    pages_term_types = pages_inventory_method("term-types")

    get_primarity = get_inventory_method("primarities")
    pages_primarities = pages_inventory_method("primarities")

    get_reliability = get_inventory_method("reliabilities")
    pages_reliabilities = pages_inventory_method("reliabilities")

    get_query_operator = get_inventory_method("query-operators")
    pages_query_operators = pages_inventory_method("query-operators")

    get_searchable_field = get_inventory_method("searchable-fields")
    pages_searchable_fields = pages_inventory_method("searchable-fields")

    def _pages(
        self, path: str, translation_language: Optional[str] = None
    ) -> Generator[InventoryElementsResponse, None, None]:
        params = {"expand": "true", "offset": 0}
        if translation_language is not None:
            params["trans_lang"] = translation_language
        page = InventoryElementsResponse.from_dict(
            self.session.get(
                path,
                params=params,
                do_auth=False,
                api_type=API.EM,
            )
        )
        yield page

        while page.next is not None:
            page = InventoryElementsResponse.from_dict(
                self.session.get_meta_resource(page.next)
            )
            yield page

Functions

def get_inventory_method(path: str) ‑> Callable[[Inventories, Optional[str], Optional[str], Optional[bool]], InventoryElement]

Expand source code

def get_inventory_method(
    path: str,
) -> Callable[
    ["Inventories", Optional[str], Optional[str], Optional[bool]], InventoryElement
]:
    def inventory_method(
        self: "Inventories",
        code: Optional[str],
        name: Optional[str],
        is_official: Optional[bool],
    ) -> InventoryElement:

        official_func = lambda l: (is_official is None) or (
            l.is_official == is_official
        )

        filter_by_name = lambda l: (l.name == name) and official_func(l)
        filter_by_code = lambda l: (l.code == code) and official_func(l)
        filter_func = filter_by_name if code is None else filter_by_code

        response = {}
        for page in self._pages(f"/inventories/_{path}"):
            for language in page.items:
                if filter_func(language):
                    response = language
        return response

    return inventory_method

def pages_inventory_method(path: str) ‑> Callable[[Inventories, Optional[str]], Generator[InventoryElementsResponse, None, None]]

Expand source code

def pages_inventory_method(
    path: str,
) -> Callable[
    ["Inventories", Optional[str]], Generator[InventoryElementsResponse, None, None]
]:
    return lambda self, translation_language=None: self._pages(
        f"/inventories/_{path}", translation_language
    )

Classes

class Inventories (session: Session)

Expand source code

class Inventories(BaseResource):
    get_language = get_inventory_method("languages")
    pages_languages = pages_inventory_method("languages")

    get_term_type = get_inventory_method("term-types")
    pages_term_types = pages_inventory_method("term-types")

    get_primarity = get_inventory_method("primarities")
    pages_primarities = pages_inventory_method("primarities")

    get_reliability = get_inventory_method("reliabilities")
    pages_reliabilities = pages_inventory_method("reliabilities")

    get_query_operator = get_inventory_method("query-operators")
    pages_query_operators = pages_inventory_method("query-operators")

    get_searchable_field = get_inventory_method("searchable-fields")
    pages_searchable_fields = pages_inventory_method("searchable-fields")

    def _pages(
        self, path: str, translation_language: Optional[str] = None
    ) -> Generator[InventoryElementsResponse, None, None]:
        params = {"expand": "true", "offset": 0}
        if translation_language is not None:
            params["trans_lang"] = translation_language
        page = InventoryElementsResponse.from_dict(
            self.session.get(
                path,
                params=params,
                do_auth=False,
                api_type=API.EM,
            )
        )
        yield page

        while page.next is not None:
            page = InventoryElementsResponse.from_dict(
                self.session.get_meta_resource(page.next)
            )
            yield page

Ancestors

BaseResource

Class variables

var session : Session

Methods

def get_language(self: Inventories, code: Optional[str], name: Optional[str], is_official: Optional[bool]) ‑> InventoryElement

Expand source code

def inventory_method(
    self: "Inventories",
    code: Optional[str],
    name: Optional[str],
    is_official: Optional[bool],
) -> InventoryElement:

    official_func = lambda l: (is_official is None) or (
        l.is_official == is_official
    )

    filter_by_name = lambda l: (l.name == name) and official_func(l)
    filter_by_code = lambda l: (l.code == code) and official_func(l)
    filter_func = filter_by_name if code is None else filter_by_code

    response = {}
    for page in self._pages(f"/inventories/_{path}"):
        for language in page.items:
            if filter_func(language):
                response = language
    return response

def get_primarity(self: Inventories, code: Optional[str], name: Optional[str], is_official: Optional[bool]) ‑> InventoryElement

Expand source code

def inventory_method(
    self: "Inventories",
    code: Optional[str],
    name: Optional[str],
    is_official: Optional[bool],
) -> InventoryElement:

    official_func = lambda l: (is_official is None) or (
        l.is_official == is_official
    )

    filter_by_name = lambda l: (l.name == name) and official_func(l)
    filter_by_code = lambda l: (l.code == code) and official_func(l)
    filter_func = filter_by_name if code is None else filter_by_code

    response = {}
    for page in self._pages(f"/inventories/_{path}"):
        for language in page.items:
            if filter_func(language):
                response = language
    return response

def get_query_operator(self: Inventories, code: Optional[str], name: Optional[str], is_official: Optional[bool]) ‑> InventoryElement

Expand source code

def inventory_method(
    self: "Inventories",
    code: Optional[str],
    name: Optional[str],
    is_official: Optional[bool],
) -> InventoryElement:

    official_func = lambda l: (is_official is None) or (
        l.is_official == is_official
    )

    filter_by_name = lambda l: (l.name == name) and official_func(l)
    filter_by_code = lambda l: (l.code == code) and official_func(l)
    filter_func = filter_by_name if code is None else filter_by_code

    response = {}
    for page in self._pages(f"/inventories/_{path}"):
        for language in page.items:
            if filter_func(language):
                response = language
    return response

def get_reliability(self: Inventories, code: Optional[str], name: Optional[str], is_official: Optional[bool]) ‑> InventoryElement

Expand source code

def inventory_method(
    self: "Inventories",
    code: Optional[str],
    name: Optional[str],
    is_official: Optional[bool],
) -> InventoryElement:

    official_func = lambda l: (is_official is None) or (
        l.is_official == is_official
    )

    filter_by_name = lambda l: (l.name == name) and official_func(l)
    filter_by_code = lambda l: (l.code == code) and official_func(l)
    filter_func = filter_by_name if code is None else filter_by_code

    response = {}
    for page in self._pages(f"/inventories/_{path}"):
        for language in page.items:
            if filter_func(language):
                response = language
    return response

def get_searchable_field(self: Inventories, code: Optional[str], name: Optional[str], is_official: Optional[bool]) ‑> InventoryElement

Expand source code

def inventory_method(
    self: "Inventories",
    code: Optional[str],
    name: Optional[str],
    is_official: Optional[bool],
) -> InventoryElement:

    official_func = lambda l: (is_official is None) or (
        l.is_official == is_official
    )

    filter_by_name = lambda l: (l.name == name) and official_func(l)
    filter_by_code = lambda l: (l.code == code) and official_func(l)
    filter_func = filter_by_name if code is None else filter_by_code

    response = {}
    for page in self._pages(f"/inventories/_{path}"):
        for language in page.items:
            if filter_func(language):
                response = language
    return response

def get_term_type(self: Inventories, code: Optional[str], name: Optional[str], is_official: Optional[bool]) ‑> InventoryElement

Expand source code

def inventory_method(
    self: "Inventories",
    code: Optional[str],
    name: Optional[str],
    is_official: Optional[bool],
) -> InventoryElement:

    official_func = lambda l: (is_official is None) or (
        l.is_official == is_official
    )

    filter_by_name = lambda l: (l.name == name) and official_func(l)
    filter_by_code = lambda l: (l.code == code) and official_func(l)
    filter_func = filter_by_name if code is None else filter_by_code

    response = {}
    for page in self._pages(f"/inventories/_{path}"):
        for language in page.items:
            if filter_func(language):
                response = language
    return response

def pages_languages(self, translation_language=None)

Expand source code

return lambda self, translation_language=None: self._pages(
    f"/inventories/_{path}", translation_language
)

def pages_primarities(self, translation_language=None)

Expand source code

return lambda self, translation_language=None: self._pages(
    f"/inventories/_{path}", translation_language
)

def pages_query_operators(self, translation_language=None)

Expand source code

return lambda self, translation_language=None: self._pages(
    f"/inventories/_{path}", translation_language
)

def pages_reliabilities(self, translation_language=None)

Expand source code

return lambda self, translation_language=None: self._pages(
    f"/inventories/_{path}", translation_language
)

def pages_searchable_fields(self, translation_language=None)

Expand source code

return lambda self, translation_language=None: self._pages(
    f"/inventories/_{path}", translation_language
)

def pages_term_types(self, translation_language=None)

Expand source code

return lambda self, translation_language=None: self._pages(
    f"/inventories/_{path}", translation_language
)

class InventoryElement (meta: Meta, code: str, name: str, is_official: Optional[bool] = None, deprecated: Optional[bool] = None)

InventoryElement(meta: piate.api.response.Meta, code: str, name: str, is_official: Optional[bool] = None, deprecated: Optional[bool] = None)

Expand source code

@dataclass_json(undefined=Undefined.RAISE)
@dataclass
class InventoryElement:
    meta: Meta
    code: str
    name: str
    is_official: Optional[bool] = field(default=None)
    deprecated: Optional[bool] = field(default=None)

    class Compact(TypedDict):
        code: str
        name: str
        deprecated: Optional[bool]
        is_official: Optional[bool]

    def compact(self) -> Compact:
        compact = {"code": self.code, "name": self.name}
        if self.is_official is not None:
            compact["is_official"] = self.is_official
        if self.deprecated:
            compact["deprecated"] = self.deprecated
        return compact

Class variables

var Compact: dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)
var code : str
var dataclass_json_config
var deprecated : Optional[bool]
var is_official : Optional[bool]
var meta : Meta
var name : str

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

def schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) ‑> dataclasses_json.mm.SchemaF[~A]

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def compact(self) ‑> InventoryElement.Compact

Expand source code

def compact(self) -> Compact:
    compact = {"code": self.code, "name": self.name}
    if self.is_official is not None:
        compact["is_official"] = self.is_official
    if self.deprecated:
        compact["deprecated"] = self.deprecated
    return compact

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

def to_json(self, *, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Union[int, str, ForwardRef(None)] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) ‑> str

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)

class InventoryElementsResponse (items: List[InventoryElement], meta: Meta, offset: Optional[int] = None, limit: Optional[int] = None, size: Optional[int] = None, next: Optional[Meta] = None)

InventoryElementsResponse(items: List[piate.api.resources.inventories.InventoryElement], meta: piate.api.response.Meta, offset: Optional[int] = None, limit: Optional[int] = None, size: Optional[int] = None, next: Optional[piate.api.response.Meta] = None)

Expand source code

@dataclass_json(undefined=Undefined.RAISE)
@dataclass
class InventoryElementsResponse:
    items: List[InventoryElement]

    meta: Meta

    offset: Optional[int] = field(default=None)
    limit: Optional[int] = field(default=None)
    size: Optional[int] = field(default=None)

    next: Optional[Meta] = field(default=None)

    def compact(self) -> List[InventoryElement.Compact]:
        return [item.compact() for item in self.items]

Class variables

var dataclass_json_config
var items : List[InventoryElement]
var limit : Optional[int]
var meta : Meta
var next : Optional[Meta]
var offset : Optional[int]
var size : Optional[int]

Static methods

def from_dict(kvs: Union[dict, list, str, int, float, bool, ForwardRef(None)], *, infer_missing=False) ‑> ~A

Expand source code

@classmethod
def from_dict(cls: Type[A],
              kvs: Json,
              *,
              infer_missing=False) -> A:
    return _decode_dataclass(cls, kvs, infer_missing)

def from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) ‑> ~A

Expand source code

@classmethod
def from_json(cls: Type[A],
              s: JsonData,
              *,
              parse_float=None,
              parse_int=None,
              parse_constant=None,
              infer_missing=False,
              **kw) -> A:
    kvs = json.loads(s,
                     parse_float=parse_float,
                     parse_int=parse_int,
                     parse_constant=parse_constant,
                     **kw)
    return cls.from_dict(kvs, infer_missing=infer_missing)

Expand source code

@classmethod
def schema(cls: Type[A],
           *,
           infer_missing: bool = False,
           only=None,
           exclude=(),
           many: bool = False,
           context=None,
           load_only=(),
           dump_only=(),
           partial: bool = False,
           unknown=None) -> SchemaType:
    Schema = build_schema(cls, DataClassJsonMixin, infer_missing, partial)

    if unknown is None:
        undefined_parameter_action = _undefined_parameter_action_safe(cls)
        if undefined_parameter_action is not None:
            # We can just make use of the same-named mm keywords
            unknown = undefined_parameter_action.name.lower()

    return Schema(only=only,
                  exclude=exclude,
                  many=many,
                  context=context,
                  load_only=load_only,
                  dump_only=dump_only,
                  partial=partial,
                  unknown=unknown)

Methods

def compact(self) ‑> List[InventoryElement.Compact]

Expand source code

def compact(self) -> List[InventoryElement.Compact]:
    return [item.compact() for item in self.items]

def to_dict(self, encode_json=False) ‑> Dict[str, Union[dict, list, str, int, float, bool, ForwardRef(None)]]

Expand source code

def to_dict(self, encode_json=False) -> Dict[str, Json]:
    return _asdict(self, encode_json=encode_json)

Expand source code

def to_json(self,
            *,
            skipkeys: bool = False,
            ensure_ascii: bool = True,
            check_circular: bool = True,
            allow_nan: bool = True,
            indent: Optional[Union[int, str]] = None,
            separators: Tuple[str, str] = None,
            default: Callable = None,
            sort_keys: bool = False,
            **kw) -> str:
    return json.dumps(self.to_dict(encode_json=False),
                      cls=_ExtendedEncoder,
                      skipkeys=skipkeys,
                      ensure_ascii=ensure_ascii,
                      check_circular=check_circular,
                      allow_nan=allow_nan,
                      indent=indent,
                      separators=separators,
                      default=default,
                      sort_keys=sort_keys,
                      **kw)