GlcoinCore/squeaknode
1
0
Fork 0
mirror of https://github.com/yzernik/squeaknode.git synced 2026-08-14 12:43:24 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Auto format (#92)

Browse source
This commit is contained in:
Jonathan Zernik 2020-07-19 02:31:45 -07:00 • committed by GitHub
parent 195e5be57f
commit 4bbca79979
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 51 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

9
itests/test.py
View file

@ -305,12 +305,13 @@ def run():
# Get a squeak display item
get_squeak_display_response = admin_stub.GetSqueakDisplay(
squeak_admin_pb2.GetSqueakDisplayRequest(
hash=make_squeak_hash,
)
squeak_admin_pb2.GetSqueakDisplayRequest(hash=make_squeak_hash,)
)
print("Get squeak display response: " + str(get_squeak_display_response))
assert get_squeak_display_response.squeak_display_entry.content_str == "Hello from the profile on the server!"
assert (
get_squeak_display_response.squeak_display_entry.content_str
== "Hello from the profile on the server!"
)
if __name__ == "__main__":

6
squeakserver/admin/squeak_admin_server_handler.py
View file

@ -44,5 +44,9 @@ class SqueakAdminServerHandler(object):
squeak_entry_with_profile = self.squeak_node.get_squeak_entry_with_profile(
squeak_hash
)
logger.info("Got squeak entry with profile for hash: {}".format(squeak_entry_with_profile))
logger.info(
"Got squeak entry with profile for hash: {}".format(
squeak_entry_with_profile
)
)
return squeak_entry_with_profile

4
squeakserver/admin/squeak_admin_server_servicer.py
View file

@ -52,7 +52,9 @@ class SqueakAdminServerServicer(squeak_admin_pb2_grpc.SqueakAdminServicer):
def GetSqueakDisplay(self, request, context):
squeak_hash = request.hash
squeak_entry_with_profile = self.handler.handle_get_squeak_display_entry(squeak_hash)
squeak_entry_with_profile = self.handler.handle_get_squeak_display_entry(
squeak_hash
)
squeak_entry = squeak_entry_with_profile.squeak_entry
squeak = squeak_entry.squeak
return squeak_admin_pb2.GetSqueakDisplayReply(

7
squeakserver/blockchain/bitcoin_blockchain_client.py
View file

@ -1,10 +1,9 @@
import logging
import json
import logging
from typing import Optional
import requests
logger = logging.getLogger(__name__)
@ -39,9 +38,7 @@ class BitcoinBlockchainClient:
"id": 0,
}
response = requests.post(
self.url,
data=json.dumps(payload),
headers=self.headers,
self.url, data=json.dumps(payload), headers=self.headers,
).json()
logger.info("Got header request result: {}".format(response))

1
squeakserver/core/squeak_entry.py
View file

@ -1,4 +1,3 @@
from collections import namedtuple
SqueakEntry = namedtuple("SqueakEntry", ["squeak", "block_header"])

5
squeakserver/core/squeak_entry_with_profile.py
View file

@ -1,4 +1,5 @@
from collections import namedtuple
SqueakEntryWithProfile = namedtuple("SqueakEntryWithProfile", ["squeak_entry", "squeak_profile"])
SqueakEntryWithProfile = namedtuple(
"SqueakEntryWithProfile", ["squeak_entry", "squeak_profile"]
)

2
squeakserver/node/squeak_block_verifier.py
View file

@ -1,8 +1,6 @@
import logging
import queue
from squeakserver.blockchain.util import parse_block_header
logger = logging.getLogger(__name__)

2
squeakserver/server/main.py
View file

@ -34,7 +34,7 @@ def load_lightning_client(config) -> LNDLightningClient:
def load_lightning_host_port(config) -> LNDLightningClient:
lnd_host = config.get("lnd","external_host", fallback=None)
lnd_host = config.get("lnd", "external_host", fallback=None)
lnd_port = int(config["lnd"]["port"])
return LightningAddressHostPort(lnd_host, lnd_port,)

65
squeakserver/server/postgres_db.py
View file

@ -3,14 +3,12 @@ from contextlib import contextmanager
from psycopg2 import pool
from psycopg2.extras import DictCursor
from squeak.core import CSqueak
from squeakserver.server.squeak_profile import SqueakProfile
from squeakserver.server.util import get_hash
from squeakserver.core.squeak_entry import SqueakEntry
from squeakserver.core.squeak_entry_with_profile import SqueakEntryWithProfile
from squeakserver.server.squeak_profile import SqueakProfile
from squeakserver.server.util import get_hash
logger = logging.getLogger(__name__)
@ -79,7 +77,7 @@ class PostgresDb:
)
# get the generated hash back
row = curs.fetchone()
return bytes.fromhex(row['hash'])
return bytes.fromhex(row["hash"])
def get_squeak_entry(self, squeak_hash):
""" Get a squeak. """
@ -110,8 +108,7 @@ class PostgresDb:
squeak_entry = self._parse_squeak_entry(row)
squeak_profile = self._parse_squeak_profile(row)
return SqueakEntryWithProfile(
squeak_entry=squeak_entry,
squeak_profile=squeak_profile,
squeak_entry=squeak_entry, squeak_profile=squeak_profile,
)
def lookup_squeaks(self, addresses, min_block, max_block):
@ -134,7 +131,7 @@ class PostgresDb:
# logger.info(curs.mogrify(sql, (addresses_tuple, min_block, max_block)))
curs.execute(sql, (addresses_tuple, min_block, max_block))
rows = curs.fetchall()
hashes = [bytes.fromhex(row['hash']) for row in rows]
hashes = [bytes.fromhex(row["hash"]) for row in rows]
return hashes
def insert_profile(self, squeak_profile):
@ -159,8 +156,8 @@ class PostgresDb:
logger.info("Inserted new profile")
# get the new profile id back
row = curs.fetchone()
logger.info("New profile id: {}".format(row['profile_id']))
return row['profile_id']
logger.info("New profile id: {}".format(row["profile_id"]))
return row["profile_id"]
def get_profile(self, profile_id):
""" Get a profile. """
@ -181,7 +178,7 @@ class PostgresDb:
with self.get_cursor() as curs:
curs.execute(sql)
rows = curs.fetchall()
hashes = [bytes.fromhex(row['hash']) for row in rows]
hashes = [bytes.fromhex(row["hash"]) for row in rows]
return hashes
def delete_squeak(self, squeak_hash):
@ -206,36 +203,38 @@ class PostgresDb:
curs.execute(sql, (block_header, squeak_hash_str,))
def _parse_squeak_entry(self, row):
vch_decryption_key_column=row['vch_decryption_key']
vch_decryption_key = bytes(vch_decryption_key_column) if vch_decryption_key_column else None
vch_decryption_key_column = row["vch_decryption_key"]
vch_decryption_key = (
bytes(vch_decryption_key_column) if vch_decryption_key_column else None
)
squeak = CSqueak(
nVersion=row['n_version'],
hashEncContent=bytes.fromhex(row['hash_enc_content']),
hashReplySqk=bytes.fromhex(row['hash_reply_sqk']),
hashBlock=bytes.fromhex(row['hash_block']),
nBlockHeight=row['n_block_height'],
vchScriptPubKey=bytes(row['vch_script_pub_key']),
vchEncryptionKey=bytes(row['vch_encryption_key']),
encDatakey=bytes.fromhex(row['enc_data_key']),
iv=bytes.fromhex((row['iv'])),
nTime=row['n_time'],
nNonce=row['n_nonce'],
encContent=bytes.fromhex((row['enc_content'])),
vchScriptSig=bytes(row['vch_script_sig']),
nVersion=row["n_version"],
hashEncContent=bytes.fromhex(row["hash_enc_content"]),
hashReplySqk=bytes.fromhex(row["hash_reply_sqk"]),
hashBlock=bytes.fromhex(row["hash_block"]),
nBlockHeight=row["n_block_height"],
vchScriptPubKey=bytes(row["vch_script_pub_key"]),
vchEncryptionKey=bytes(row["vch_encryption_key"]),
encDatakey=bytes.fromhex(row["enc_data_key"]),
iv=bytes.fromhex((row["iv"])),
nTime=row["n_time"],
nNonce=row["n_nonce"],
encContent=bytes.fromhex((row["enc_content"])),
vchScriptSig=bytes(row["vch_script_sig"]),
vchDecryptionKey=vch_decryption_key,
)
block_header_column = row['block_header']
block_header_column = row["block_header"]
block_header = bytes(block_header_column) if block_header_column else None
return SqueakEntry(squeak=squeak, block_header=block_header)
def _parse_squeak_profile(self, row):
private_key_column = row['private_key']
private_key_column = row["private_key"]
private_key = bytes(private_key_column) if private_key_column else None
return SqueakProfile(
profile_id=row['profile_id'],
profile_name=row['profile_name'],
profile_id=row["profile_id"],
profile_name=row["profile_name"],
private_key=private_key,
address=row['address'],
sharing=row['sharing'],
following=row['following'],
address=row["address"],
sharing=row["sharing"],
following=row["following"],
)