fix: broken imports
This commit is contained in:
@@ -4,8 +4,6 @@ from __future__ import annotations
|
||||
import os
|
||||
import json
|
||||
import logging
|
||||
import base64
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
# external imports
|
||||
from cic_types import MetadataPointer
|
||||
@@ -14,8 +12,6 @@ from hexathon import strip_0x
|
||||
|
||||
# local imports
|
||||
from cic.contract.base import Data, data_dir
|
||||
from cic.writers import OutputWriter
|
||||
from cic_types.ext.metadata import MetadataRequestsHandler
|
||||
from cic.utils import object_to_str
|
||||
|
||||
logg = logging.getLogger(__name__)
|
||||
@@ -139,25 +135,3 @@ class Meta(Data):
|
||||
def __str__(self):
|
||||
return object_to_str(self, ["name", "contact", "country_code", "location"])
|
||||
|
||||
|
||||
class MetadataWriter(OutputWriter):
|
||||
"""Custom writer for publishing data under immutable content-addressed pointers in the cic-meta storage backend.
|
||||
|
||||
Data that is not utf-8 will be converted to base64 before publishing.
|
||||
|
||||
Implements cic.writers.OutputWriter
|
||||
"""
|
||||
|
||||
def write(self, k, v):
|
||||
rq = MetadataRequestsHandler(MetadataPointer.NONE, bytes.fromhex(k))
|
||||
try:
|
||||
v = v.decode("utf-8")
|
||||
v = json.loads(v)
|
||||
logg.debug(f"metadatawriter bindecode {k} {v}")
|
||||
except UnicodeDecodeError:
|
||||
v = base64.b64encode(v).decode("utf-8")
|
||||
v = json.loads(json.dumps(v, separators=(",", ":")))
|
||||
logg.debug(f"metadatawriter b64encode {k} {v}")
|
||||
r = rq.create(v)
|
||||
logg.info(f"metadata submitted at {k}")
|
||||
return r
|
||||
|
||||
@@ -7,7 +7,7 @@ import logging
|
||||
from chainlib.chain import ChainSpec
|
||||
|
||||
# local imports
|
||||
from cic.contract.components.base import Data, data_dir
|
||||
from cic.contract.base import Data, data_dir
|
||||
|
||||
logg = logging.getLogger(__name__)
|
||||
|
||||
@@ -35,9 +35,8 @@ class Network(Data):
|
||||
"""
|
||||
super(Network, self).load()
|
||||
|
||||
f = open(self.network_path, 'r')
|
||||
o = json.load(f)
|
||||
f.close()
|
||||
with open(self.network_path, 'r', encoding='utf-8') as f:
|
||||
o = json.load(f)
|
||||
|
||||
self.resources = o['resources']
|
||||
|
||||
@@ -53,9 +52,8 @@ class Network(Data):
|
||||
|
||||
network_template_file_path = os.path.join(data_dir, f'network_template_v{self.version()}.json')
|
||||
|
||||
f = open(network_template_file_path)
|
||||
o_part = json.load(f)
|
||||
f.close()
|
||||
with open(network_template_file_path, encoding='utf-8') as f:
|
||||
o_part = json.load(f)
|
||||
|
||||
self.resources = {}
|
||||
for v in self.targets:
|
||||
@@ -67,11 +65,10 @@ class Network(Data):
|
||||
def save(self):
|
||||
"""Save network settings to file.
|
||||
"""
|
||||
f = open(self.network_path, 'w')
|
||||
json.dump({
|
||||
'resources': self.resources,
|
||||
}, f, sort_keys=True, indent="\t")
|
||||
f.close()
|
||||
with open(self.network_path, 'w', encoding='utf-8') as f:
|
||||
json.dump({
|
||||
'resources': self.resources,
|
||||
}, f, sort_keys=True, indent="\t")
|
||||
|
||||
|
||||
def resource(self, k):
|
||||
@@ -83,8 +80,8 @@ class Network(Data):
|
||||
:return: Extension settings
|
||||
"""
|
||||
v = self.resources.get(k)
|
||||
if v == None:
|
||||
raise AttributeError('no defined reference for {}'.format(k))
|
||||
if v is None:
|
||||
raise AttributeError(f'No defined reference for {k}')
|
||||
return v
|
||||
|
||||
|
||||
@@ -129,7 +126,7 @@ class Network(Data):
|
||||
"""
|
||||
chain_spec_dict = chain_spec.asdict()
|
||||
for k in chain_spec_dict.keys():
|
||||
logg.debug('resources {}'.format(self.resources))
|
||||
logg.debug(f'resources: {self.resources}')
|
||||
self.resources[resource_key]['chain_spec'][k] = chain_spec_dict[k]
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user