"""Extract information from the file system, such as filesize. """ # standard imports import os import typing # bsie imports from bsie.utils import bsfs, node, ns # inner-module imports from .. import base # exports __all__: typing.Sequence[str] = ( 'Stat', ) ## code ## class Stat(base.Extractor): """Extract information from the file system.""" CONTENT_READER = 'bsie.reader.stat.Stat' # mapping from predicate to handler function. _callmap: typing.Dict[bsfs.schema.Predicate, typing.Callable[[os.stat_result], typing.Any]] def __init__(self): super().__init__(bsfs.schema.from_string(base.SCHEMA_PREAMBLE + ''' bse:filesize rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsn:Entity ; rdfs:range xsd:integer ; rdfs:label "File size"^^xsd:string ; schema:description "File size of entity in some filesystem."^^xsd:string ; bsfs:unique "true"^^xsd:boolean . ''')) self._callmap = { self.schema.predicate(ns.bse.filesize): self.__filesize, } def extract( self, subject: node.Node, content: os.stat_result, principals: typing.Iterable[bsfs.schema.Predicate], ) -> typing.Iterator[typing.Tuple[node.Node, bsfs.schema.Predicate, typing.Any]]: for pred in principals: # find callback clbk = self._callmap.get(pred) if clbk is None: continue # get value value = clbk(content) if value is None: continue # produce triple yield subject, pred, value def __filesize(self, content: os.stat_result) -> typing.Optional[int]: """Return the file size.""" try: return content.st_size except Exception: # pylint: disable=broad-except # we explicitly want to catch everything # FIXME: some kind of error reporting (e.g. logging) return None ## EOF ##