Module signals_notebook.entities.contentful_entity
Expand source code
import abc
import cgi
import json
import logging
from enum import Enum
from typing import Any, List, Optional
from signals_notebook.api import SignalsNotebookApi
from signals_notebook.common_types import EntityType, File
from signals_notebook.entities import Entity
from signals_notebook.entities.container import Container
from signals_notebook.jinja_env import env
from signals_notebook.utils import FSHandler
log = logging.getLogger(__name__)
class ContentfulEntity(Entity, abc.ABC):
class ContentType(str, Enum):
BYTES = 'application/octet-stream'
def get_content(self) -> File:
raise NotImplementedError
@classmethod
@abc.abstractmethod
def _get_entity_type(cls) -> EntityType:
pass
@classmethod
def create(
cls,
*,
container: Container,
name: str,
content_type: str = ContentType.BYTES,
content: bytes = b'',
force: bool = True,
) -> Entity:
raise NotImplementedError
def _get_content(self, format: Optional[str] = None) -> File:
api = SignalsNotebookApi.get_default_api()
log.debug('Get content for: %s| %s', self.__class__.__name__, self.eid)
response = api.call(
method='GET',
path=(self._get_endpoint(), self.eid, 'export'),
params={
'format': format,
},
)
content_disposition = response.headers.get('content-disposition', '')
_, params = cgi.parse_header(content_disposition)
return File(
name=params['filename'],
content=response.content,
content_type=response.headers.get('content-type'),
)
def get_html(self) -> str:
"""Get in HTML format
Returns:
Rendered template as a string
"""
file = self._get_content()
data = {
'name': self.name,
'content': 'data:{};base64,{}'.format(file.content_type, file.base64.decode('ascii')),
}
template = env.get_template(self._template_name)
log.info('Html template for %s:%s has been rendered.', self.__class__.__name__, self.eid)
return template.render(data=data)
def dump(self, base_path: str, fs_handler: FSHandler, alias: Optional[List[str]] = None) -> None:
"""Dump ContentfulEntity entity
Args:
base_path: content path where create dump
fs_handler: FSHandler
alias: Backup alias
Returns:
"""
content = self.get_content()
metadata = {
'file_name': content.name,
'content_type': content.content_type,
**{k: v for k, v in self.dict().items() if k in ('name', 'description', 'eid')},
}
fs_handler.write(
fs_handler.join_path(base_path, self.eid, 'metadata.json'),
json.dumps(metadata),
base_alias=alias + [self.name, '__Metadata'] if alias else None,
)
file_name = content.name
data = content.content
fs_handler.write(
fs_handler.join_path(base_path, self.eid, file_name),
data,
base_alias=alias + [self.name, file_name] if alias else None,
)
@classmethod
def load(cls, path: str, fs_handler: FSHandler, parent: Container) -> None:
"""Load ContentfulEntity entity
Args:
path: content path
fs_handler: FSHandler
parent: Container where load ContentfulEntity entity
Returns:
"""
cls._load(path, fs_handler, parent)
@classmethod
def _load(cls, path: str, fs_handler: FSHandler, parent: Any) -> None:
metadata_path = fs_handler.join_path(path, 'metadata.json')
metadata = json.loads(fs_handler.read(metadata_path))
content_path = fs_handler.join_path(path, metadata['file_name'])
content_type = metadata.get('content_type')
content = fs_handler.read(content_path)
if content_type:
cls.create(container=parent, name=metadata['name'], content=content, content_type=content_type, force=True)
else:
cls.create(container=parent, name=metadata['name'], content=content, force=True)
Classes
class ContentfulEntity (**data: Any)
-
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 ContentfulEntity(Entity, abc.ABC): class ContentType(str, Enum): BYTES = 'application/octet-stream' def get_content(self) -> File: raise NotImplementedError @classmethod @abc.abstractmethod def _get_entity_type(cls) -> EntityType: pass @classmethod def create( cls, *, container: Container, name: str, content_type: str = ContentType.BYTES, content: bytes = b'', force: bool = True, ) -> Entity: raise NotImplementedError def _get_content(self, format: Optional[str] = None) -> File: api = SignalsNotebookApi.get_default_api() log.debug('Get content for: %s| %s', self.__class__.__name__, self.eid) response = api.call( method='GET', path=(self._get_endpoint(), self.eid, 'export'), params={ 'format': format, }, ) content_disposition = response.headers.get('content-disposition', '') _, params = cgi.parse_header(content_disposition) return File( name=params['filename'], content=response.content, content_type=response.headers.get('content-type'), ) def get_html(self) -> str: """Get in HTML format Returns: Rendered template as a string """ file = self._get_content() data = { 'name': self.name, 'content': 'data:{};base64,{}'.format(file.content_type, file.base64.decode('ascii')), } template = env.get_template(self._template_name) log.info('Html template for %s:%s has been rendered.', self.__class__.__name__, self.eid) return template.render(data=data) def dump(self, base_path: str, fs_handler: FSHandler, alias: Optional[List[str]] = None) -> None: """Dump ContentfulEntity entity Args: base_path: content path where create dump fs_handler: FSHandler alias: Backup alias Returns: """ content = self.get_content() metadata = { 'file_name': content.name, 'content_type': content.content_type, **{k: v for k, v in self.dict().items() if k in ('name', 'description', 'eid')}, } fs_handler.write( fs_handler.join_path(base_path, self.eid, 'metadata.json'), json.dumps(metadata), base_alias=alias + [self.name, '__Metadata'] if alias else None, ) file_name = content.name data = content.content fs_handler.write( fs_handler.join_path(base_path, self.eid, file_name), data, base_alias=alias + [self.name, file_name] if alias else None, ) @classmethod def load(cls, path: str, fs_handler: FSHandler, parent: Container) -> None: """Load ContentfulEntity entity Args: path: content path fs_handler: FSHandler parent: Container where load ContentfulEntity entity Returns: """ cls._load(path, fs_handler, parent) @classmethod def _load(cls, path: str, fs_handler: FSHandler, parent: Any) -> None: metadata_path = fs_handler.join_path(path, 'metadata.json') metadata = json.loads(fs_handler.read(metadata_path)) content_path = fs_handler.join_path(path, metadata['file_name']) content_type = metadata.get('content_type') content = fs_handler.read(content_path) if content_type: cls.create(container=parent, name=metadata['name'], content=content, content_type=content_type, force=True) else: cls.create(container=parent, name=metadata['name'], content=content, force=True)
Ancestors
- Entity
- pydantic.main.BaseModel
- pydantic.utils.Representation
- abc.ABC
Subclasses
- BiologicalSequence
- ChemicalDrawing
- Excel
- Image
- MaterialsTable
- SubExperimentLayout
- SubExperimentSummary
- PlateContainer
- PowerPoint
- SamplesContainer
- Spotfire
- Text
- TodoList
- UploadedResource
- Word
Class variables
var ContentType
-
An enumeration.
var created_at : datetime.datetime
var description : Optional[str]
var digest : Optional[str]
var edited_at : datetime.datetime
var eid : EID
var name : str
var type : str
Static methods
def create(*, container: Container, name: str, content_type: str = ContentType.BYTES, content: bytes = b'', force: bool = True) ‑> Entity
-
Expand source code
@classmethod def create( cls, *, container: Container, name: str, content_type: str = ContentType.BYTES, content: bytes = b'', force: bool = True, ) -> Entity: raise NotImplementedError
def load(path: str, fs_handler: FSHandler, parent: Container) ‑> None
-
Load ContentfulEntity entity
Args
path
- content path
fs_handler
- FSHandler
parent
- Container where load ContentfulEntity entity
Returns:
Expand source code
@classmethod def load(cls, path: str, fs_handler: FSHandler, parent: Container) -> None: """Load ContentfulEntity entity Args: path: content path fs_handler: FSHandler parent: Container where load ContentfulEntity entity Returns: """ cls._load(path, fs_handler, parent)
Methods
def dump(self, base_path: str, fs_handler: FSHandler, alias: Optional[List[str]] = None) ‑> None
-
Dump ContentfulEntity entity
Args
base_path
- content path where create dump
fs_handler
- FSHandler
alias
- Backup alias
Returns:
Expand source code
def dump(self, base_path: str, fs_handler: FSHandler, alias: Optional[List[str]] = None) -> None: """Dump ContentfulEntity entity Args: base_path: content path where create dump fs_handler: FSHandler alias: Backup alias Returns: """ content = self.get_content() metadata = { 'file_name': content.name, 'content_type': content.content_type, **{k: v for k, v in self.dict().items() if k in ('name', 'description', 'eid')}, } fs_handler.write( fs_handler.join_path(base_path, self.eid, 'metadata.json'), json.dumps(metadata), base_alias=alias + [self.name, '__Metadata'] if alias else None, ) file_name = content.name data = content.content fs_handler.write( fs_handler.join_path(base_path, self.eid, file_name), data, base_alias=alias + [self.name, file_name] if alias else None, )
def get_content(self) ‑> File
-
Expand source code
def get_content(self) -> File: raise NotImplementedError
def get_html(self) ‑> str
-
Get in HTML format
Returns
Rendered template as a string
Expand source code
def get_html(self) -> str: """Get in HTML format Returns: Rendered template as a string """ file = self._get_content() data = { 'name': self.name, 'content': 'data:{};base64,{}'.format(file.content_type, file.base64.decode('ascii')), } template = env.get_template(self._template_name) log.info('Html template for %s:%s has been rendered.', self.__class__.__name__, self.eid) return template.render(data=data)
Inherited members