Higher-order function to handle storage endpoint and
return a singledispatch generic function with preset storage strategies.
This is a form of generic function dispatch where the
implementation is chosen based on the type of a single argument.
Usage:
store = storage.ipfs() # default localhost:5001
stored_object = store(b'test bytes') # auto-choose the storage strategy
Parameters:
| Name |
Type |
Description |
Default |
endpoint |
Optional[str]
|
Endpoint to connect to the API. If the endpoint is not specified, localhost is used instead. |
None
|
Returns:
| Type |
Description |
Store
|
Singledispatch decorated function |
Source code in nucleus/sdk/storage/store.py
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133 | def ipfs(endpoint: Optional[str] = None) -> Store:
"""Higher-order function to handle storage endpoint and
return a singledispatch generic function with preset storage strategies.
This is a form of generic function dispatch where the
implementation is chosen based on the type of a single argument.
Usage:
store = storage.ipfs() # default localhost:5001
stored_object = store(b'test bytes') # auto-choose the storage strategy
:param endpoint: Endpoint to connect to the API. If the endpoint is not specified, localhost is used instead.
:return: Singledispatch decorated function
"""
# Connect to the IPFS API interface
api = ipfs_.rpc(endpoint)
@functools.singledispatch
def store(data: Storable) -> Object:
"""Storage single dispatch factory.
Uses the data input type to infer the right storage strategy.
:param data: The model to dispatch
:return: Object instance
"""
raise NotImplementedError(f'cannot process not registered storable `{data}')
@store.register
def _(data: FileType) -> Object:
"""Store files in IPFS.
:param data: File to store
:return: Object instance
"""
command = Add(File(data.path))
# expected /add output from API
# {Hash: .., Name: .., Size: ...}
file_output = api(command)
return Object(
name=file_output['Name'],
hash=CID(file_output['Hash']),
size=int(file_output['Size']),
)
@store.register
def _(data: Path) -> Object:
"""Store directory in IPFS.
:param data: Directory path to store
:return: Object instance
"""
command = Add(
input=Dir(data),
wrap_with_directory=True,
)
# expected /add output from API
# {Hash: .., Name: .., Size: ...}
dir_output = api(command)
return Object(
name=dir_output['Name'],
hash=CID(dir_output['Hash']),
size=int(dir_output['Size']),
)
@store.register
def _(data: bytes) -> Object:
"""Store bytes in IPFS.
Store bytes in raw blocks.
:param data: Bytes to store
:return: Object instance
"""
command = BlockPut(Text(data))
# expected block/put output from API
# {Key: .., Size: ..}
output = api(command)
return Object(
name=output['Key'],
hash=CID(output['Key']),
size=len(data),
)
@store.register
def _(data: str) -> Object:
"""String string in IPFS.
Encode string to bytes and store it in raw blocks.
:param data: String to store
:return: Object instance
"""
bytes_ = bytes(data, 'utf-8')
return store(bytes_)
@store.register
def _(data: JSON) -> Object:
"""Store JSON in IPFS Dag.
:param data: JSON to store
:return: Object instance
"""
bytes_ = bytes(data)
command = DagPut(Text(bytes_))
# expected block/put output from API
# {"Cid": { "/": "<cid-string>" }}
output = api(command)
raw_cid = output['Cid']['/']
return Object(
name=raw_cid,
hash=CID(raw_cid),
size=len(data),
)
return store
|