import re
import threading
import warnings
from pycoin.networks.default import get_current_netcode
from .bitcoind import BitcoindProvider
from .blockexplorer import BlockExplorerProvider
from .blockchain_info import BlockchainInfoProvider
from .blockcypher import BlockcypherProvider
from .chain_so import ChainSoProvider
from .insight import InsightProvider
from .btgexp import BTGExpProvider
from .env import main_cache_dir, config_string_for_netcode_from_env
from .env import tx_read_cache_dirs, tx_writable_cache_dir
from .tx_db import TxDb
THREAD_LOCALS = threading.local()
# PYCOIN_BTC_PROVIDERS="blockchain.info blockexplorer.com blockcypher.com chain.so"
# PYCOIN_BTC_PROVIDERS="insight:http(s?)://hostname/url bitcoinrpc://user:passwd@hostname:8332"
[docs]def service_provider_methods(method_name, service_providers):
methods = [getattr(m, method_name, None) for m in service_providers]
methods = [m for m in methods if m]
return methods
[docs]def spendables_for_address(address, netcode, format=None):
"""
Return a list of Spendable objects for the
given bitcoin address.
Set format to "text" or "dict" to transform return value
from an object to a string or dict.
This is intended to be a convenience function. There is no way to know that
the list returned is a complete list of spendables for the address in question.
You can verify that they really do come from the existing transaction
by calling tx_utils.validate_unspents.
"""
if format:
method = "as_%s" % format
for m in service_provider_methods("spendables_for_address", get_default_providers_for_netcode(netcode)):
try:
spendables = m(address)
if format:
spendables = [getattr(s, method)() for s in spendables]
return spendables
except Exception:
pass
return []
[docs]def get_tx_db(netcode=None):
lookup_methods = service_provider_methods("tx_for_tx_hash", get_default_providers_for_netcode(netcode))
read_cache_dirs = tx_read_cache_dirs()
writable_cache_dir = tx_writable_cache_dir()
return TxDb(lookup_methods=lookup_methods, read_only_paths=read_cache_dirs,
writable_cache_path=writable_cache_dir)
[docs]def message_about_tx_cache_env():
if main_cache_dir() is None:
return "consider setting environment variable PYCOIN_CACHE_DIR=~/.pycoin_cache to"\
" cache transactions fetched via web services"
[docs]def all_providers_message(method, netcode):
if len(service_provider_methods(method, get_default_providers_for_netcode(netcode))) == 0:
return "no service providers found for %s; consider setting environment variable "\
"PYCOIN_%s_PROVIDERS" % (method, netcode)
[docs]def message_about_spendables_for_address_env(netcode):
return all_providers_message("spendables_for_address", netcode)
[docs]def message_about_tx_for_tx_hash_env(netcode):
return all_providers_message("tx_for_tx_hash", netcode)
[docs]def bitcoin_rpc_init(match, netcode):
username, password, hostname, port = match.group("user", "password", "hostname", "port")
return BitcoindProvider("http://%s:%s@%s:%s" % (username, password, hostname, port))
[docs]def insight_init(match, netcode):
return InsightProvider(base_url=match.group("url"), netcode=netcode)
DESCRIPTOR_CRE_INIT_TUPLES = [
(re.compile(
r"^bitcoinrpc://(?P<user>\S*):(?P<password>\S*)\@(?P<hostname>\S*)(:(?P<port>\d*))"),
bitcoin_rpc_init),
(re.compile(r"^blockchain\.info$"), lambda m, netcode: BlockchainInfoProvider(netcode)),
(re.compile(r"^blockcypher\.com$"), lambda m, netcode: BlockcypherProvider(netcode)),
(re.compile(r"^blockexplorer\.com$"), lambda m, netcode: BlockExplorerProvider(netcode)),
(re.compile(r"^chain\.so$"), lambda m, netcode: ChainSoProvider(netcode)),
(re.compile(r"^insight:(?P<url>\S*)$"), insight_init),
(re.compile(r"^btgexp.com"), lambda m, netcode: BTGExpProvider()),
]
[docs]def provider_for_descriptor_and_netcode(descriptor, netcode=None):
if netcode is None:
netcode = get_current_netcode()
for cre, f in DESCRIPTOR_CRE_INIT_TUPLES:
m = cre.match(descriptor)
if m:
return f(m, netcode)
return None
[docs]def providers_for_config_string(config_string, netcode):
providers = []
for d in config_string.split():
p = provider_for_descriptor_and_netcode(d, netcode)
if p:
providers.append(p)
else:
warnings.warn("can't parse provider %s in config string" % d)
return providers
[docs]def providers_for_netcode_from_env(netcode):
return providers_for_config_string(config_string_for_netcode_from_env(netcode), netcode)
[docs]def get_default_providers_for_netcode(netcode=None):
if netcode is None:
netcode = get_current_netcode()
if not hasattr(THREAD_LOCALS, "providers"):
THREAD_LOCALS.providers = {}
if netcode not in THREAD_LOCALS.providers:
THREAD_LOCALS.providers[netcode] = providers_for_netcode_from_env(netcode)
return THREAD_LOCALS.providers[netcode]
[docs]def set_default_providers_for_netcode(netcode, provider_list):
if not hasattr(THREAD_LOCALS, "providers"):
THREAD_LOCALS.providers = {}
THREAD_LOCALS.providers[netcode] = provider_list