1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
"""
Part of the bsie module.
A copy of the license is provided with the project.
Author: Matthias Baumgartner, 2022
"""
# standard imports
import abc
import os
import typing
# 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.bsfs.File:
return self.name_file(node)
raise errors.ProgrammingError('no naming policy available for {node.node_type}')
def name_file(self, node: Node) -> Node:
"""Set a bsfs:File 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 = (self._prefix + 'file')[fragment]
return node
## EOF ##
|
