Source code for pycoin.services.tx_db
import os.path
from pycoin.coins.bitcoin.Tx import Tx
from pycoin.encoding.hexbytes import b2h_rev
[docs]class TxDb(object):
"""
This object can be used in many places that expect a dict.
"""
def __init__(self, lookup_methods=[], read_only_paths=[], writable_cache_path=None):
self.lookup_methods = lookup_methods
self.read_only_paths = read_only_paths
if writable_cache_path:
self.read_only_paths.append(writable_cache_path)
self.writable_cache_path = writable_cache_path
if self.writable_cache_path and not os.path.exists(self.writable_cache_path):
os.makedirs(self.writable_cache_path)
[docs] def paths_for_hash(self, hash):
name = b2h_rev(hash)
for base_dir in self.read_only_paths:
p = os.path.join(base_dir, "%s_tx.bin" % name)
if os.path.exists(p):
yield p
[docs] def put(self, tx):
name = b2h_rev(tx.hash())
if self.writable_cache_path:
try:
path = os.path.join(self.writable_cache_path, "%s_tx.bin" % name)
with open(path, "wb") as f:
tx.stream(f)
except IOError:
pass
[docs] def get(self, key):
for path in self.paths_for_hash(key):
try:
tx = Tx.parse(open(path, "rb"))
if tx and tx.hash() == key:
return tx
except IOError:
pass
for method in self.lookup_methods:
try:
tx = method(key)
if tx and tx.hash() == key:
self.put(tx)
return tx
except Exception:
pass
return None
def __getitem__(self, key):
raise NotImplemented
def __setitem__(self, key, val):
if val.hash() != key:
raise ValueError("bad key %s for %s" % (b2h_rev(key), val))
self.put(val)