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

Add more type hints controller (#843)

Browse source 
* Use more type hints in controller

* Add type hints for squeak entry detail and sent offer
This commit is contained in:
Jonathan Zernik 2021-02-13 09:09:04 -08:00 • committed by GitHub
parent 00c8bd9769
commit 4f1272cc26
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 105 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

25
itests/tests/test_squeak_node.py
View file

@ -682,7 +682,10 @@ def test_delete_profile(server_stub, admin_stub, contact_profile_id):
profile_id=contact_profile_id,
)
)
assert "Profile not found." in str(excinfo.value)
assert (
"Profile not found with id: {}.".format(contact_profile_id)
in str(excinfo.value)
)
def test_get_profile_private_key(server_stub, admin_stub, signing_profile_id):
@ -862,7 +865,10 @@ def test_delete_peer(server_stub, admin_stub, peer_id):
peer_id=peer_id,
)
)
assert "Peer not found." in str(excinfo.value)
assert (
"Peer with id {} not found.".format(peer_id)
in str(excinfo.value)
)
def test_list_channels(server_stub, admin_stub, lightning_client, saved_squeak_hash):
@ -1290,13 +1296,18 @@ def test_download_single_squeak(
)
# Get the squeak display item (should be empty)
get_squeak_display_response = other_admin_stub.GetSqueakDisplay(
squeak_admin_pb2.GetSqueakDisplayRequest(
squeak_hash=saved_squeak_hash,
with pytest.raises(Exception) as excinfo:
get_squeak_display_response = other_admin_stub.GetSqueakDisplay(
squeak_admin_pb2.GetSqueakDisplayRequest(
squeak_hash=saved_squeak_hash,
)
)
assert (
"Squeak not found with hash: {}.".format(saved_squeak_hash)
in str(excinfo.value)
)
assert get_squeak_display_response.squeak_display_entry.squeak_hash == ""
# Get the buy offer (should be empty)
# Get buy offers for the squeak hash (should be empty)
get_buy_offers_response = other_admin_stub.GetBuyOffers(
squeak_admin_pb2.GetBuyOffersRequest(
squeak_hash=saved_squeak_hash,

28
squeaknode/admin/squeak_admin_server_handler.py
View file

@ -144,8 +144,6 @@ class SqueakAdminServerHandler(object):
profile_id = request.profile_id
logger.info("Handle get squeak profile with id: {}".format(profile_id))
squeak_profile = self.squeak_controller.get_squeak_profile(profile_id)
if squeak_profile is None:
raise Exception("Profile not found.")
squeak_profile_msg = squeak_profile_to_message(squeak_profile)
return squeak_admin_pb2.GetSqueakProfileReply(
squeak_profile=squeak_profile_msg,
@ -157,8 +155,6 @@ class SqueakAdminServerHandler(object):
"Handle get squeak profile with address: {}".format(address))
squeak_profile = self.squeak_controller.get_squeak_profile_by_address(
address)
if squeak_profile is None:
raise Exception("Profile not found.")
squeak_profile_msg = squeak_profile_to_message(squeak_profile)
return squeak_admin_pb2.GetSqueakProfileByAddressReply(
squeak_profile=squeak_profile_msg
@ -169,8 +165,6 @@ class SqueakAdminServerHandler(object):
logger.info("Handle get squeak profile with name: {}".format(name))
squeak_profile = self.squeak_controller.get_squeak_profile_by_name(
name)
if squeak_profile is None:
raise Exception("Profile not found.")
squeak_profile_msg = squeak_profile_to_message(squeak_profile)
return squeak_admin_pb2.GetSqueakProfileByNameReply(
squeak_profile=squeak_profile_msg
@ -275,13 +269,12 @@ class SqueakAdminServerHandler(object):
logger.info(
"Handle get squeak display entry for hash: {}".format(squeak_hash_str))
squeak_entry_with_profile = (
self.squeak_controller.get_squeak_entry_with_profile(squeak_hash)
self.squeak_controller.get_squeak_entry_with_profile(
squeak_hash
)
)
if squeak_entry_with_profile is None:
display_message = None
else:
display_message = squeak_entry_to_message(
squeak_entry_with_profile)
display_message = squeak_entry_to_message(
squeak_entry_with_profile)
return squeak_admin_pb2.GetSqueakDisplayReply(
squeak_display_entry=display_message
)
@ -411,8 +404,6 @@ class SqueakAdminServerHandler(object):
peer_id = request.peer_id
logger.info("Handle get squeak peer with id: {}".format(peer_id))
squeak_peer = self.squeak_controller.get_peer(peer_id)
if squeak_peer is None:
raise Exception("Peer not found.")
squeak_peer_msg = squeak_peer_to_message(squeak_peer)
return squeak_admin_pb2.GetPeerReply(
squeak_peer=squeak_peer_msg,
@ -422,7 +413,8 @@ class SqueakAdminServerHandler(object):
logger.info("Handle get squeak peers")
squeak_peers = self.squeak_controller.get_peers()
squeak_peer_msgs = [
squeak_peer_to_message(squeak_peer) for squeak_peer in squeak_peers
squeak_peer_to_message(squeak_peer)
for squeak_peer in squeak_peers
]
return squeak_admin_pb2.GetPeersReply(
squeak_peers=squeak_peer_msgs,
@ -485,8 +477,6 @@ class SqueakAdminServerHandler(object):
offer_id = request.offer_id
logger.info("Handle get buy offer for hash: {}".format(offer_id))
offer = self.squeak_controller.get_buy_offer_with_peer(offer_id)
if offer is None:
raise Exception("Offer not found.")
offer_msg = offer_entry_to_message(offer)
return squeak_admin_pb2.GetBuyOfferReply(
offer=offer_msg,
@ -537,8 +527,6 @@ class SqueakAdminServerHandler(object):
logger.info(
"Handle get sent payment with id: {}".format(sent_payment_id))
sent_payment = self.squeak_controller.get_sent_payment(sent_payment_id)
if sent_payment is None:
raise Exception("SentPayment not found.")
sent_payment_msg = sent_payment_with_peer_to_message(sent_payment)
return squeak_admin_pb2.GetSentPaymentReply(
sent_payment=sent_payment_msg,
@ -554,8 +542,6 @@ class SqueakAdminServerHandler(object):
squeak_hash
)
)
if squeak_entry_with_profile is None:
raise Exception("Squeak details not found.")
detail_message = squeak_entry_to_detail_message(
squeak_entry_with_profile)
return squeak_admin_pb2.GetSqueakDetailsReply(

14
squeaknode/db/squeak_db.py
View file

@ -488,7 +488,7 @@ class SqueakDb:
return None
return self._parse_squeak_profile(row)
def set_profile_following(self, profile_id: int, following: bool):
def set_profile_following(self, profile_id: int, following: bool) -> None:
""" Set a profile is following. """
stmt = (
self.profiles.update()
@ -498,7 +498,7 @@ class SqueakDb:
with self.get_connection() as connection:
connection.execute(stmt)
def set_profile_sharing(self, profile_id: int, sharing: bool):
def set_profile_sharing(self, profile_id: int, sharing: bool) -> None:
""" Set a profile is sharing. """
stmt = (
self.profiles.update()
@ -508,7 +508,7 @@ class SqueakDb:
with self.get_connection() as connection:
connection.execute(stmt)
def set_profile_name(self, profile_id: int, profile_name: str):
def set_profile_name(self, profile_id: int, profile_name: str) -> None:
""" Set a profile name. """
stmt = (
self.profiles.update()
@ -518,7 +518,7 @@ class SqueakDb:
with self.get_connection() as connection:
connection.execute(stmt)
def delete_profile(self, profile_id: int):
def delete_profile(self, profile_id: int) -> None:
""" Delete a profile. """
delete_profile_stmt = self.profiles.delete().where(
self.profiles.c.profile_id == profile_id
@ -526,7 +526,7 @@ class SqueakDb:
with self.get_connection() as connection:
connection.execute(delete_profile_stmt)
def set_profile_image(self, profile_id: int, profile_image: bytes):
def set_profile_image(self, profile_id: int, profile_image: bytes) -> None:
""" Set a profile image. """
stmt = (
self.profiles.update()
@ -536,7 +536,7 @@ class SqueakDb:
with self.get_connection() as connection:
connection.execute(stmt)
def set_squeak_decryption_key(self, squeak_hash: bytes, secret_key: bytes):
def set_squeak_decryption_key(self, squeak_hash: bytes, secret_key: bytes) -> None:
""" Set the decryption key of a squeak. """
stmt = (
self.squeaks.update()
@ -546,7 +546,7 @@ class SqueakDb:
with self.get_connection() as connection:
connection.execute(stmt)
def delete_squeak(self, squeak_hash: bytes):
def delete_squeak(self, squeak_hash: bytes) -> None:
""" Delete a squeak. """
delete_squeak_stmt = self.squeaks.delete().where(
self.squeaks.c.hash == squeak_hash.hex()

108
squeaknode/node/squeak_controller.py
View file

@ -10,9 +10,12 @@ from squeak.core.signing import CSqueakAddress
from squeaknode.core.block_range import BlockRange
from squeaknode.core.offer import Offer
from squeaknode.core.received_offer import ReceivedOffer
from squeaknode.core.received_offer_with_peer import ReceivedOfferWithPeer
from squeaknode.core.received_payment_summary import ReceivedPaymentSummary
from squeaknode.core.sent_offer import SentOffer
from squeaknode.core.sent_payment_summary import SentPaymentSummary
from squeaknode.core.sent_payment_with_peer import SentPaymentWithPeer
from squeaknode.core.squeak_entry_with_profile import SqueakEntryWithProfile
from squeaknode.core.squeak_peer import SqueakPeer
from squeaknode.core.squeak_profile import SqueakProfile
from squeaknode.core.util import get_hash
@ -26,6 +29,7 @@ logger = logging.getLogger(__name__)
class SqueakController:
def __init__(
self,
squeak_db,
@ -220,40 +224,55 @@ class SqueakController:
)
return self.squeak_db.insert_profile(squeak_profile)
def get_signing_profiles(self):
def get_signing_profiles(self) -> List[SqueakProfile]:
return self.squeak_db.get_signing_profiles()
def get_contact_profiles(self):
def get_contact_profiles(self) -> List[SqueakProfile]:
return self.squeak_db.get_contact_profiles()
def get_squeak_profile(self, profile_id: int):
return self.squeak_db.get_profile(profile_id)
def get_squeak_profile(self, profile_id: int) -> SqueakProfile:
profile = self.squeak_db.get_profile(profile_id)
if profile is None:
raise Exception("Profile not found with id: {}.".format(
profile_id,
))
return profile
def get_squeak_profile_by_address(self, address: str):
return self.squeak_db.get_profile_by_address(address)
def get_squeak_profile_by_address(self, address: str) -> SqueakProfile:
profile = self.squeak_db.get_profile_by_address(address)
if profile is None:
raise Exception("Profile not found with address: {}.".format(
address,
))
return profile
def get_squeak_profile_by_name(self, name: str):
return self.squeak_db.get_profile_by_name(name)
def get_squeak_profile_by_name(self, name: str) -> SqueakProfile:
profile = self.squeak_db.get_profile_by_name(name)
if profile is None:
raise Exception("Profile not found with name: {}.".format(
name,
))
return profile
def set_squeak_profile_following(self, profile_id: int, following: bool):
def set_squeak_profile_following(self, profile_id: int, following: bool) -> None:
self.squeak_db.set_profile_following(profile_id, following)
def set_squeak_profile_sharing(self, profile_id: int, sharing: bool):
def set_squeak_profile_sharing(self, profile_id: int, sharing: bool) -> None:
self.squeak_db.set_profile_sharing(profile_id, sharing)
def rename_squeak_profile(self, profile_id: int, profile_name: str):
def rename_squeak_profile(self, profile_id: int, profile_name: str) -> None:
self.squeak_db.set_profile_name(profile_id, profile_name)
def delete_squeak_profile(self, profile_id: int):
def delete_squeak_profile(self, profile_id: int) -> None:
self.squeak_db.delete_profile(profile_id)
def set_squeak_profile_image(self, profile_id: int, profile_image: bytes):
def set_squeak_profile_image(self, profile_id: int, profile_image: bytes) -> None:
self.squeak_db.set_profile_image(profile_id, profile_image)
def clear_squeak_profile_image(self, profile_id: int):
def clear_squeak_profile_image(self, profile_id: int) -> None:
self.squeak_db.set_profile_image(profile_id, None)
def get_squeak_profile_private_key(self, profile_id: int):
def get_squeak_profile_private_key(self, profile_id: int) -> bytes:
profile = self.get_squeak_profile(profile_id)
if profile.private_key is None:
raise Exception("Profile with id: {} does not have a private key.".format(
@ -261,20 +280,17 @@ class SqueakController:
))
return profile.private_key
def make_squeak(self, profile_id: int, content_str: str, replyto_hash: bytes):
def make_squeak(self, profile_id: int, content_str: str, replyto_hash: bytes) -> bytes:
squeak_profile = self.squeak_db.get_profile(profile_id)
squeak_entry = self.squeak_core.make_squeak(
squeak_profile, content_str, replyto_hash)
return self.save_created_squeak(squeak_entry.squeak)
# inserted_squeak_hash = self.squeak_db.insert_squeak(
# squeak_entry.squeak, squeak_entry.block_header)
# return inserted_squeak_hash
def delete_squeak(self, squeak_hash: bytes):
def delete_squeak(self, squeak_hash: bytes) -> None:
num_deleted_offers = self.squeak_db.delete_offers_for_squeak(
squeak_hash)
logger.info("Deleted number of offers : {}".format(num_deleted_offers))
return self.squeak_db.delete_squeak(squeak_hash)
self.squeak_db.delete_squeak(squeak_hash)
def create_peer(self, peer_name: str, host: str, port: int):
if len(peer_name) == 0:
@ -292,8 +308,13 @@ class SqueakController:
)
return self.squeak_db.insert_peer(squeak_peer)
def get_peer(self, peer_id: int):
return self.squeak_db.get_peer(peer_id)
def get_peer(self, peer_id: int) -> SqueakPeer:
peer = self.squeak_db.get_peer(peer_id)
if peer is None:
raise Exception("Peer with id {} not found.".format(
peer_id,
))
return peer
def get_peers(self):
return self.squeak_db.get_peers()
@ -319,8 +340,14 @@ class SqueakController:
def get_buy_offers_with_peer(self, squeak_hash: bytes):
return self.squeak_db.get_offers_with_peer(squeak_hash)
def get_buy_offer_with_peer(self, offer_id: int):
return self.squeak_db.get_offer_with_peer(offer_id)
def get_buy_offer_with_peer(self, received_offer_id: int) -> ReceivedOfferWithPeer:
received_offer_with_peer = self.squeak_db.get_offer_with_peer(
received_offer_id)
if received_offer_with_peer is None:
raise Exception("Received offer with id {} not found.".format(
received_offer_id,
))
return received_offer_with_peer
def pay_offer(self, received_offer_id: int) -> int:
# Get the offer from the database
@ -359,11 +386,16 @@ class SqueakController:
secret_key,
)
def get_sent_payments(self):
def get_sent_payments(self) -> List[SentPaymentWithPeer]:
return self.squeak_db.get_sent_payments()
def get_sent_payment(self, sent_payment_id: int):
return self.squeak_db.get_sent_payment(sent_payment_id)
def get_sent_payment(self, sent_payment_id: int) -> SentPaymentWithPeer:
sent_payment = self.squeak_db.get_sent_payment(sent_payment_id)
if sent_payment is None:
raise Exception("Sent payment not found with id: {}.".format(
sent_payment_id,
))
return sent_payment
def get_sent_offers(self):
return self.squeak_db.get_sent_offers()
@ -395,20 +427,26 @@ class SqueakController:
for payment in client.get_received_payments():
yield payment
def get_block_range(self):
def get_block_range(self) -> BlockRange:
max_block = self.squeak_core.get_best_block_height()
block_interval = self.config.sync.block_interval
min_block = max(0, max_block - block_interval)
return BlockRange(min_block, max_block)
def get_network(self):
def get_network(self) -> str:
return self.config.core.network
def get_offer(self, squeak: CSqueak, offer: Offer, peer: SqueakPeer) -> ReceivedOffer:
return self.squeak_core.unpack_offer(squeak, offer, peer)
def get_squeak_entry_with_profile(self, squeak_hash: bytes):
return self.squeak_db.get_squeak_entry_with_profile(squeak_hash)
def get_squeak_entry_with_profile(self, squeak_hash: bytes) -> SqueakEntryWithProfile:
squeak_entry_with_profile = self.squeak_db.get_squeak_entry_with_profile(
squeak_hash)
if squeak_entry_with_profile is None:
raise Exception("Squeak not found with hash: {}.".format(
squeak_hash.hex(),
))
return squeak_entry_with_profile
def get_timeline_squeak_entries_with_profile(self):
return self.squeak_db.get_timeline_squeak_entries_with_profile()
@ -455,15 +493,15 @@ class SqueakController:
peer_id,
)
def save_offer(self, received_offer: ReceivedOffer):
def save_offer(self, received_offer: ReceivedOffer) -> None:
logger.info("Saving received offer: {}".format(received_offer))
self.squeak_db.insert_received_offer(received_offer)
def get_followed_addresses(self):
def get_followed_addresses(self) -> List[str]:
followed_profiles = self.squeak_db.get_following_profiles()
return [profile.address for profile in followed_profiles]
def get_sharing_addresses(self):
def get_sharing_addresses(self) -> List[str]:
sharing_profiles = self.squeak_db.get_sharing_profiles()
return [profile.address for profile in sharing_profiles]