aboutsummaryrefslogtreecommitdiffstats
path: root/bsie/lib/naming_policy.py
diff options
context:
space:
mode:
Diffstat (limited to 'bsie/lib/naming_policy.py')
-rw-r--r--bsie/lib/naming_policy.py141
1 files changed, 0 insertions, 141 deletions
diff --git a/bsie/lib/naming_policy.py b/bsie/lib/naming_policy.py
deleted file mode 100644
index fbdbeb0..0000000
--- a/bsie/lib/naming_policy.py
+++ /dev/null
@@ -1,141 +0,0 @@
-
-# standard imports
-import abc
-import os
-import typing
-
-# external imports
-import urllib.parse
-
-# bsie imports
-from bsie.utils import bsfs, errors, ns
-from bsie.utils.node import Node
-
-# exports
-__all__: typing.Sequence[str] = (
- 'DefaultNamingPolicy',
- )
-
-
-## code ##
-
-class NamingPolicy():
- """Determine node uri's from node hints."""
- def __call__(
- self,
- iterable: typing.Iterable[typing.Tuple[Node, bsfs.URI, typing.Any]],
- ):
- """Apply the policy on a triple iterator."""
- return NamingPolicyIterator(self, iterable)
-
- @abc.abstractmethod
- def handle_node(self, node: Node) -> Node:
- """Apply the policy on a node."""
-
-
-class NamingPolicyIterator():
- """Iterates over triples, determines uris according to a *policy* as it goes."""
-
- # source triple iterator.
- _iterable: typing.Iterable[typing.Tuple[Node, bsfs.URI, typing.Any]]
-
- # naming policy
- _policy: NamingPolicy
-
- def __init__(
- self,
- policy: NamingPolicy,
- iterable: typing.Iterable[typing.Tuple[Node, bsfs.URI, typing.Any]],
- ):
- self._iterable = iterable
- self._policy = policy
-
- def __iter__(self):
- for node, pred, value in self._iterable:
- # handle subject
- self._policy.handle_node(node)
- # handle value
- if isinstance(value, Node):
- self._policy.handle_node(value)
- # yield triple
- yield node, pred, value
-
-
-class DefaultNamingPolicy(NamingPolicy):
- """Compose URIs as <host/user/node_type#fragment>
-
- What information is used as fragment depends on the node type.
- Typically, the default is to use the "ucid" hint.
- The fallback in all cases is to generate a random uuid.
-
- Never changes previously assigned uris. Sets uris in-place.
-
- """
-
- def __init__(
- self,
- host: bsfs.URI,
- user: str,
- ):
- self._prefix = bsfs.Namespace(os.path.join(host, user))
- self._uuid = bsfs.uuid.UUID()
-
- def handle_node(self, node: Node) -> Node:
- if node.uri is not None:
- return node
- if node.node_type == ns.bsn.Entity:
- return self.name_entity(node)
- if node.node_type == ns.bsn.Preview:
- return self.name_preview(node)
- if node.node_type == ns.bsn.Tag:
- return self.name_tag(node)
- if node.node_type == ns.bsn.Face:
- return self.name_face(node)
- raise errors.ProgrammingError(f'no naming policy available for {node.node_type}')
-
- def name_entity(self, node: Node) -> Node:
- """Set a bsn:Entity node's uri fragment to its ucid."""
- if 'ucid' in node.hints: # content id
- fragment = node.hints['ucid']
- else: # random name
- fragment = self._uuid()
- node.uri = getattr(self._prefix.file(), fragment)
- return node
-
- def name_preview(self, node: Node) -> Node:
- """Set a bsn:Preview node's uri fragment to its ucid.
- Uses its source fragment as fallback. Appends the size if provided.
- """
- fragment = None
- if 'ucid' in node.hints: # content id
- fragment = node.hints['ucid']
- if fragment is None and 'source' in node.hints: # source id
- self.handle_node(node.hints['source'])
- fragment = node.hints['source'].uri.get('fragment', None)
- if fragment is None: # random name
- fragment = self._uuid()
- if 'size' in node.hints: # append size
- fragment += '_s' + str(node.hints['size'])
- node.uri = getattr(self._prefix.preview(), fragment)
- return node
-
- def name_tag(self, node: Node) -> Node:
- # NOTE: Must ensure to produce the same name for that tags with the same label.
- if 'label' in node.hints: # tag label
- fragment = urllib.parse.quote(node.hints['label'])
- else: # random name
- fragment = self._uuid()
- # FIXME: match to existing tags in bsfs storage!
- node.uri = getattr(self._prefix.tag(), fragment)
- return node
-
- def name_face(self, node: Node) -> Node:
- if 'ucid' in node.hints: # content id
- fragment = node.hints['ucid']
- else: # random name
- fragment = self._uuid()
- node.uri = getattr(self._prefix.face(), fragment)
- return node
-
-
-## EOF ##