Module signals_notebook.attributes.attribute
Expand source code
import logging
from enum import Enum
from typing import cast, Generator, Literal, Optional
from pydantic import BaseModel
from pydantic.fields import PrivateAttr
from signals_notebook.api import SignalsNotebookApi
from signals_notebook.common_types import AttrID, ObjectType, Response, ResponseData
log = logging.getLogger(__name__)
class Action(str, Enum):
UPDATE = 'update'
DELETE = 'delete'
CREATE = 'create'
class _Attributes(BaseModel):
action: Action
value: Optional[str]
class _OptionRepresentation(BaseModel):
id: Optional[str]
type: Literal[ObjectType.ATTRIBUTE_OPTION] = ObjectType.ATTRIBUTE_OPTION
attributes: _Attributes
class _AttributeOption(BaseModel):
value: str
class _AttributeOptionResponse(Response[_AttributeOption]):
pass
class Attribute(BaseModel):
type: str
id: AttrID
name: str
_options: list[str] = PrivateAttr(default=[])
def __init__(self, *args, **kwargs):
self._options = kwargs.pop('options', [])
super().__init__(*args, **kwargs)
self.type = ObjectType.CHOICE
def __iter__(self):
if not self._options:
self._reload_options()
return self._options.__iter__()
def __len__(self) -> int:
if not self._options:
self._reload_options()
return len(self._options)
@classmethod
def _get_endpoint(cls) -> str:
return 'attributes'
@classmethod
def get(cls, id: AttrID) -> 'Attribute':
"""Get Attribute object by id
Args:
id: AttrID
Returns:
Attribute
"""
api = SignalsNotebookApi.get_default_api()
response = api.call(
method='GET',
path=(cls._get_endpoint(), id),
)
result = AttributeResponse(**response.json())
log.debug('Get Attribute with ID: %s', id)
return cast(ResponseData, result.data).body
@classmethod
def get_list(cls) -> Generator['Attribute', None, None]:
"""Get all Attributes.
Returns:
list of available Attributes
"""
api = SignalsNotebookApi.get_default_api()
log.debug('Get List of Attributes')
response = api.call(
method='GET',
path=(cls._get_endpoint(),),
)
result = AttributeResponse(**response.json())
yield from [cast(ResponseData, item).body for item in result.data]
while result.links and result.links.next:
response = api.call(
method='GET',
path=result.links.next,
)
result = AttributeResponse(**response.json())
yield from [cast(ResponseData, item).body for item in result.data]
log.debug('List of Attributes was got successfully.')
@classmethod
def create(
cls,
name: str,
description: str,
options: list[str] = None,
) -> 'Attribute':
"""Create new Attribute
Args:
name: name of new Attribute
description: description of new Attribute
options: list of available id options for created Attribute
Returns:
Attribute
"""
api = SignalsNotebookApi.get_default_api()
log.debug('Create Attribute: %s...', cls.__name__)
response = api.call(
method='POST',
path=(cls._get_endpoint(),),
json={
'data': {
'type': ObjectType.ATTRIBUTE,
'attributes': {
'name': name,
'type': ObjectType.CHOICE,
'description': description,
'options': options or [],
},
}
},
)
log.debug('Attribute: %s was created.', cls.__name__)
result = AttributeResponse(**response.json())
return cast(ResponseData, result.data).body
def delete(self) -> None:
"""Delete an Attribute by id.
Returns:
"""
api = SignalsNotebookApi.get_default_api()
log.debug('Delete Attribute: %s...', self.id)
api.call(
method='DELETE',
path=(self._get_endpoint(), self.id),
)
log.debug('Attribute: %s was deleted successfully', self.id)
def add_option(self, option: str) -> None:
"""Update option of Attribute by id.
Args:
option: option which will be added to option list in Attribute
Returns:
"""
new_option = _OptionRepresentation(attributes=_Attributes(action=Action.CREATE, value=option))
log.debug('Creating Option: %s...', self.id)
self._patch_options(new_option)
def delete_option(self, option: str) -> None:
"""Delete option of Attribute by id.
Args:
option: option which will be deleted from Attribute's option list
Returns:
"""
deleted_option = _OptionRepresentation(id=option, attributes=_Attributes(action=Action.DELETE))
log.debug('Deleting Option: %s...', self.id)
self._patch_options(deleted_option)
def update_option(self, old_option: str, new_option: str) -> None:
"""Update option of Attribute by id.
Args:
old_option: old option which will be updated
new_option: new option value
Returns:
"""
option = _OptionRepresentation(id=old_option, attributes=_Attributes(action=Action.UPDATE, value=new_option))
log.debug('Patching Option: %s...', self.id)
self._patch_options(option)
def _patch_options(self, option: _OptionRepresentation) -> None:
api = SignalsNotebookApi.get_default_api()
api.call(
method='PATCH',
path=(self._get_endpoint(), self.id, 'options'),
json={'data': [option.dict(exclude_none=True)]},
)
self._reload_options()
log.debug('Attribute: %s was reloaded successfully', self.id)
def _reload_options(self):
self._options = []
api = SignalsNotebookApi.get_default_api()
response = api.call(
method='GET',
path=(self._get_endpoint(), self.id, 'options'),
)
result = _AttributeOptionResponse(**response.json())
for item in result.data:
body = cast(ResponseData, item).body
option = cast(_AttributeOption, body)
self._options.append(option.value)
@property
def options(self) -> list[str]:
"""Get Attribute options
Returns:
list[str]
"""
if not self._options:
self._reload_options()
return self._options
class AttributeResponse(Response[Attribute]):
pass
Classes
class Action (value, names=None, *, module=None, qualname=None, type=None, start=1)
-
An enumeration.
Expand source code
class Action(str, Enum): UPDATE = 'update' DELETE = 'delete' CREATE = 'create'
Ancestors
- builtins.str
- enum.Enum
Class variables
var CREATE
var DELETE
var UPDATE
class Attribute (*args, **kwargs)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class Attribute(BaseModel): type: str id: AttrID name: str _options: list[str] = PrivateAttr(default=[]) def __init__(self, *args, **kwargs): self._options = kwargs.pop('options', []) super().__init__(*args, **kwargs) self.type = ObjectType.CHOICE def __iter__(self): if not self._options: self._reload_options() return self._options.__iter__() def __len__(self) -> int: if not self._options: self._reload_options() return len(self._options) @classmethod def _get_endpoint(cls) -> str: return 'attributes' @classmethod def get(cls, id: AttrID) -> 'Attribute': """Get Attribute object by id Args: id: AttrID Returns: Attribute """ api = SignalsNotebookApi.get_default_api() response = api.call( method='GET', path=(cls._get_endpoint(), id), ) result = AttributeResponse(**response.json()) log.debug('Get Attribute with ID: %s', id) return cast(ResponseData, result.data).body @classmethod def get_list(cls) -> Generator['Attribute', None, None]: """Get all Attributes. Returns: list of available Attributes """ api = SignalsNotebookApi.get_default_api() log.debug('Get List of Attributes') response = api.call( method='GET', path=(cls._get_endpoint(),), ) result = AttributeResponse(**response.json()) yield from [cast(ResponseData, item).body for item in result.data] while result.links and result.links.next: response = api.call( method='GET', path=result.links.next, ) result = AttributeResponse(**response.json()) yield from [cast(ResponseData, item).body for item in result.data] log.debug('List of Attributes was got successfully.') @classmethod def create( cls, name: str, description: str, options: list[str] = None, ) -> 'Attribute': """Create new Attribute Args: name: name of new Attribute description: description of new Attribute options: list of available id options for created Attribute Returns: Attribute """ api = SignalsNotebookApi.get_default_api() log.debug('Create Attribute: %s...', cls.__name__) response = api.call( method='POST', path=(cls._get_endpoint(),), json={ 'data': { 'type': ObjectType.ATTRIBUTE, 'attributes': { 'name': name, 'type': ObjectType.CHOICE, 'description': description, 'options': options or [], }, } }, ) log.debug('Attribute: %s was created.', cls.__name__) result = AttributeResponse(**response.json()) return cast(ResponseData, result.data).body def delete(self) -> None: """Delete an Attribute by id. Returns: """ api = SignalsNotebookApi.get_default_api() log.debug('Delete Attribute: %s...', self.id) api.call( method='DELETE', path=(self._get_endpoint(), self.id), ) log.debug('Attribute: %s was deleted successfully', self.id) def add_option(self, option: str) -> None: """Update option of Attribute by id. Args: option: option which will be added to option list in Attribute Returns: """ new_option = _OptionRepresentation(attributes=_Attributes(action=Action.CREATE, value=option)) log.debug('Creating Option: %s...', self.id) self._patch_options(new_option) def delete_option(self, option: str) -> None: """Delete option of Attribute by id. Args: option: option which will be deleted from Attribute's option list Returns: """ deleted_option = _OptionRepresentation(id=option, attributes=_Attributes(action=Action.DELETE)) log.debug('Deleting Option: %s...', self.id) self._patch_options(deleted_option) def update_option(self, old_option: str, new_option: str) -> None: """Update option of Attribute by id. Args: old_option: old option which will be updated new_option: new option value Returns: """ option = _OptionRepresentation(id=old_option, attributes=_Attributes(action=Action.UPDATE, value=new_option)) log.debug('Patching Option: %s...', self.id) self._patch_options(option) def _patch_options(self, option: _OptionRepresentation) -> None: api = SignalsNotebookApi.get_default_api() api.call( method='PATCH', path=(self._get_endpoint(), self.id, 'options'), json={'data': [option.dict(exclude_none=True)]}, ) self._reload_options() log.debug('Attribute: %s was reloaded successfully', self.id) def _reload_options(self): self._options = [] api = SignalsNotebookApi.get_default_api() response = api.call( method='GET', path=(self._get_endpoint(), self.id, 'options'), ) result = _AttributeOptionResponse(**response.json()) for item in result.data: body = cast(ResponseData, item).body option = cast(_AttributeOption, body) self._options.append(option.value) @property def options(self) -> list[str]: """Get Attribute options Returns: list[str] """ if not self._options: self._reload_options() return self._options
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var id : AttrID
var name : str
var type : str
Static methods
def create(name: str, description: str, options: list = None) ‑> Attribute
-
Create new Attribute
Args
name
- name of new Attribute
description
- description of new Attribute
options
- list of available id options for created Attribute
Returns
Attribute
Expand source code
@classmethod def create( cls, name: str, description: str, options: list[str] = None, ) -> 'Attribute': """Create new Attribute Args: name: name of new Attribute description: description of new Attribute options: list of available id options for created Attribute Returns: Attribute """ api = SignalsNotebookApi.get_default_api() log.debug('Create Attribute: %s...', cls.__name__) response = api.call( method='POST', path=(cls._get_endpoint(),), json={ 'data': { 'type': ObjectType.ATTRIBUTE, 'attributes': { 'name': name, 'type': ObjectType.CHOICE, 'description': description, 'options': options or [], }, } }, ) log.debug('Attribute: %s was created.', cls.__name__) result = AttributeResponse(**response.json()) return cast(ResponseData, result.data).body
def get(id: AttrID) ‑> Attribute
-
Get Attribute object by id
Args
id
- AttrID
Returns
Attribute
Expand source code
@classmethod def get(cls, id: AttrID) -> 'Attribute': """Get Attribute object by id Args: id: AttrID Returns: Attribute """ api = SignalsNotebookApi.get_default_api() response = api.call( method='GET', path=(cls._get_endpoint(), id), ) result = AttributeResponse(**response.json()) log.debug('Get Attribute with ID: %s', id) return cast(ResponseData, result.data).body
def get_list() ‑> Generator[Attribute, None, None]
-
Get all Attributes.
Returns
list of available Attributes
Expand source code
@classmethod def get_list(cls) -> Generator['Attribute', None, None]: """Get all Attributes. Returns: list of available Attributes """ api = SignalsNotebookApi.get_default_api() log.debug('Get List of Attributes') response = api.call( method='GET', path=(cls._get_endpoint(),), ) result = AttributeResponse(**response.json()) yield from [cast(ResponseData, item).body for item in result.data] while result.links and result.links.next: response = api.call( method='GET', path=result.links.next, ) result = AttributeResponse(**response.json()) yield from [cast(ResponseData, item).body for item in result.data] log.debug('List of Attributes was got successfully.')
Instance variables
var options : list
-
Get Attribute options
Returns
list[str]
Expand source code
@property def options(self) -> list[str]: """Get Attribute options Returns: list[str] """ if not self._options: self._reload_options() return self._options
Methods
def add_option(self, option: str) ‑> None
-
Update option of Attribute by id.
Args
option
- option which will be added to option list in Attribute
Returns:
Expand source code
def add_option(self, option: str) -> None: """Update option of Attribute by id. Args: option: option which will be added to option list in Attribute Returns: """ new_option = _OptionRepresentation(attributes=_Attributes(action=Action.CREATE, value=option)) log.debug('Creating Option: %s...', self.id) self._patch_options(new_option)
def delete(self) ‑> None
-
Delete an Attribute by id.
Returns:
Expand source code
def delete(self) -> None: """Delete an Attribute by id. Returns: """ api = SignalsNotebookApi.get_default_api() log.debug('Delete Attribute: %s...', self.id) api.call( method='DELETE', path=(self._get_endpoint(), self.id), ) log.debug('Attribute: %s was deleted successfully', self.id)
def delete_option(self, option: str) ‑> None
-
Delete option of Attribute by id.
Args
option
- option which will be deleted from Attribute's option list
Returns:
Expand source code
def delete_option(self, option: str) -> None: """Delete option of Attribute by id. Args: option: option which will be deleted from Attribute's option list Returns: """ deleted_option = _OptionRepresentation(id=option, attributes=_Attributes(action=Action.DELETE)) log.debug('Deleting Option: %s...', self.id) self._patch_options(deleted_option)
def update_option(self, old_option: str, new_option: str) ‑> None
-
Update option of Attribute by id.
Args
old_option
- old option which will be updated
new_option
- new option value
Returns:
Expand source code
def update_option(self, old_option: str, new_option: str) -> None: """Update option of Attribute by id. Args: old_option: old option which will be updated new_option: new option value Returns: """ option = _OptionRepresentation(id=old_option, attributes=_Attributes(action=Action.UPDATE, value=new_option)) log.debug('Patching Option: %s...', self.id) self._patch_options(option)
class AttributeResponse (**kwargs)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Expand source code
class AttributeResponse(Response[Attribute]): pass
Ancestors
- Response[Attribute]
- Response
- pydantic.generics.GenericModel
- pydantic.main.BaseModel
- pydantic.utils.Representation
- typing.Generic
Class variables
var data : Union[pydantic.generics.ResponseData[Attribute], List[pydantic.generics.ResponseData[Attribute]]]
var links : Optional[Links]
class Response[Attribute] (**kwargs)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Ancestors
- Response
- pydantic.generics.GenericModel
- pydantic.main.BaseModel
- pydantic.utils.Representation
- typing.Generic
Subclasses
Class variables
var Config
var data : Union[pydantic.generics.ResponseData[Attribute], List[pydantic.generics.ResponseData[Attribute]]]
var links : Optional[Links]
class Response[_AttributeOption] (**kwargs)
-
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
Ancestors
- Response
- pydantic.generics.GenericModel
- pydantic.main.BaseModel
- pydantic.utils.Representation
- typing.Generic
Subclasses
- signals_notebook.attributes.attribute._AttributeOptionResponse
Class variables
var Config
var data : Union[pydantic.generics.ResponseData[_AttributeOption], List[pydantic.generics.ResponseData[_AttributeOption]]]
var links : Optional[Links]