Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions Makefile
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
lint:
@ruff check . --fix
@ruff format .
.PHONY: lint
35 changes: 19 additions & 16 deletions examples/calculator.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,20 @@
from snet import sdk

config = sdk.config.Config(private_key="YOUR_PRIVATE_KEY",
eth_rpc_endpoint=f"https://sepolia.infura.io/v3/YOUR_INFURA_KEY",
concurrency=False,
force_update=False)

operators = {
"+": "add",
"-": "sub",
"*": "mul",
"/": "div"
}
config = sdk.config.Config(
private_key="YOUR_PRIVATE_KEY",
eth_rpc_endpoint="https://sepolia.infura.io/v3/YOUR_INFURA_KEY",
concurrency=False,
force_update=False,
)

operators = {"+": "add", "-": "sub", "*": "mul", "/": "div"}

snet_sdk = sdk.SnetSDK(config)
calc_client = snet_sdk.create_service_client(org_id="26072b8b6a0e448180f8c0e702ab6d2f",
service_id="Exampleservice", group_name="default_group")
calc_client = snet_sdk.create_service_client(
org_id="26072b8b6a0e448180f8c0e702ab6d2f",
service_id="Exampleservice",
group_name="default_group",
)


def parse_expression(expression):
Expand All @@ -23,12 +23,16 @@
raise Exception(f"Invalid expression '{expression}'. Three items required.")

if elements[1] not in ["+", "-", "*", "/"]:
raise Exception(f"Invalid expression '{expression}'. Operation must be '+' or '-' or '*' or '/'.")
raise Exception(

Check warning on line 26 in examples/calculator.py

View check run for this annotation

snet-sonarqube-app / SonarQube Code Analysis

examples/calculator.py#L26

Replace this generic exception class with a more specific one.
f"Invalid expression '{expression}'. Operation must be '+' or '-' or '*' or '/'."
)
try:
a = float(elements[0])
b = float(elements[2])
except ValueError:
raise Exception(f"Invalid expression '{expression}'. Operands must be integers or floating point numbers.")
raise Exception(

Check warning on line 33 in examples/calculator.py

View check run for this annotation

snet-sonarqube-app / SonarQube Code Analysis

examples/calculator.py#L33

Replace this generic exception class with a more specific one.
f"Invalid expression '{expression}'. Operands must be integers or floating point numbers."
)
op = elements[1]

return a, b, op
Expand All @@ -54,4 +58,3 @@

if __name__ == "__main__":
main()

184 changes: 131 additions & 53 deletions examples/console_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
transactions and service calls. So, to run the application, you will need to change the values in 'config'.
"""


from snet import sdk


Expand All @@ -17,7 +16,7 @@
The list is got from the MPE contract using 'get_organization_list'.
"""
print("Organizations:")
print(*map(lambda x: '\t' + x, snet_sdk.get_organization_list()), sep="\n")
print(*map(lambda x: "\t" + x, snet_sdk.get_organization_list()), sep="\n")


def list_services_for_org():
Expand All @@ -28,7 +27,10 @@
"""
org_id = input("Enter organization id: ").strip()
print("Services:")
print(*map(lambda x: '\t' + x, snet_sdk.get_services_list(org_id=org_id)), sep="\n")
print(
*map(lambda x: "\t" + x, snet_sdk.get_services_list(org_id=org_id)),
sep="\n",
)


def create_service_client():
Expand All @@ -41,7 +43,9 @@
service_id = input("Enter service id: ").strip()
group_name = input("Enter payment group name: ").strip()

service = snet_sdk.create_service_client(org_id=org_id, service_id=service_id, group_name=group_name)
service = snet_sdk.create_service_client(
org_id=org_id, service_id=service_id, group_name=group_name
)
initialized_services.append(service)

global active_service
Expand All @@ -57,7 +61,7 @@
global active_commands
print("Available commands:")
for command in active_commands.items():
print(f'\t{command[0]} - {command[1][1]}')
print(f"\t{command[0]} - {command[1][1]}")


def list_initialized_services():
Expand Down Expand Up @@ -94,8 +98,10 @@
"""
global active_service
if active_service is None:
print("No initialized services!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service.")
print(
"No initialized services!\n"

Check failure on line 102 in examples/console_app.py

View check run for this annotation

snet-sonarqube-app / SonarQube Code Analysis

examples/console_app.py#L102

Define a constant instead of duplicating this literal "No initialized services!\n" 4 times.
"Please enter 'service' to go to the service menu and then enter 'add' to add a service."
)
return None

method_name = input("Enter method name: ")
Expand Down Expand Up @@ -135,8 +141,10 @@
"""
global active_service
if active_service is None:
print("No initialized services!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service.")
print(
"No initialized services!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service."
)
return None
print(active_service.get_services_and_messages_info_as_pretty_string())

Expand All @@ -146,7 +154,9 @@
The function, which is called when the user enters the command 'balance' in the main menu.
Prints the balances of FET and MPE. It gets the balances using 'balance_of' and 'escrow_balance'.
"""
account_balance = snet_sdk.account.token_contract.functions.balanceOf(snet_sdk.account.address).call()
account_balance = snet_sdk.account.token_contract.functions.balanceOf(
snet_sdk.account.address
).call()
escrow_balance = snet_sdk.account.escrow_balance()

print(f"FET balance: {account_balance}")
Expand All @@ -168,8 +178,10 @@
Prints the current block number. It gets the block number using 'get_current_block_number'.
"""
if active_service is None:
print("No initialized services!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service.")
print(
"No initialized services!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service."
)
return None
print("Current block number: ", active_service.get_current_block_number())

Expand All @@ -183,13 +195,18 @@
to work, so there is a warning for the user about this at the beginning.
"""
if active_service is None:
print("No initialized services!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service.")
print(
"No initialized services!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service."
)
return None

is_continue = input("""Updating the channel list makes sense if the channel data has changed through other entry points.
is_continue = (
input("""Updating the channel list makes sense if the channel data has changed through other entry points.
This procedure may take several minutes.
Continue? (y/n): """).strip() == 'y'
Continue? (y/n): """).strip()
== "y"
)
if not is_continue:
return None

Expand All @@ -200,7 +217,14 @@
for service in initialized_services:
load_channels = service.load_open_channels()
for channel in load_channels:
channels.append((channel, service.org_id, service.service_id, service.group['group_name']))
channels.append(
(
channel,
service.org_id,
service.service_id,
service.group["group_name"],
)
)

print("Channels updated! Enter 'list' to print the updated list.")

Expand All @@ -214,13 +238,19 @@
"""
global active_service
global channels
additions = False
if active_service is None:
print("No initialized services! The channel can only be opened for the service!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service.")
print(
"No initialized services! The channel can only be opened for the service!\n"
"Please enter 'service' to go to the service menu and then enter 'add' to add a service."
)
return None
else:
is_continue = input("The new channel will be opened for the active service. Continue? (y/n): ").strip() == 'y'
is_continue = (
input(
"The new channel will be opened for the active service. Continue? (y/n): "
).strip()
== "y"
)
if not is_continue:
return None

Expand All @@ -230,7 +260,12 @@
is_deposit = False
if balance < amount:
print(f"Insufficient balance!\n\tCurrent MPE balance: {balance}\n\tAmount to put: {amount}")
is_deposit = input("Would you like to deposit needed amount of FET tokens in advance? (y/n): ").strip() == 'y'
is_deposit = (
input(
"Would you like to deposit needed amount of FET tokens in advance? (y/n): "
).strip()
== "y"
)
if not is_deposit:
print("Channel is not opened!")
return None
Expand All @@ -241,7 +276,14 @@
channel = active_service.open_channel(amount=amount, expiration=expiration)
else:
channel = active_service.deposit_and_open_channel(amount=amount, expiration=expiration)
channels.append((channel, active_service.org_id, active_service.service_id, active_service.group['group_name']))
channels.append(
(
channel,
active_service.org_id,
active_service.service_id,
active_service.group["group_name"],
)
)


def list_channels():
Expand All @@ -253,12 +295,14 @@
print("ORGANIZATION_ID SERVICE_ID GROUP_NAME CHANNEL_ID AMOUNT EXPIRATION")
for channel in channels:
channel[0].sync_state()
print(channel[1],
channel[2],
channel[3],
channel[0].channel_id,
channel[0].state['available_amount'],
channel[0].state['expiration'])
print(
channel[1],
channel[2],
channel[3],
channel[0].channel_id,
channel[0].state["available_amount"],
channel[0].state["expiration"],
)


def add_funds():
Expand All @@ -271,7 +315,9 @@
exists = False
for channel in channels:
if channel[0].channel_id == channel_id:
amount = int(input("Enter amount of FET tokens in cogs to add to the channel: ").strip())
amount = int(
input("Enter amount of FET tokens in cogs to add to the channel: ").strip()
)
channel[0].add_funds(amount)
exists = True
if not exists:
Expand Down Expand Up @@ -304,10 +350,12 @@
SDK configuration that is configured by the application provider.
To run the application you need to change the 'private_key', 'eth_rpc_endpoint' and 'identity_name' values.
"""
config = sdk.config.Config(private_key="YOUR_PRIVATE_KEY",
eth_rpc_endpoint=f"https://sepolia.infura.io/v3/YOUR_INFURA_KEY",
concurrency=False,
force_update=False)
config = sdk.config.Config(
private_key="YOUR_PRIVATE_KEY",
eth_rpc_endpoint="https://sepolia.infura.io/v3/YOUR_INFURA_KEY",
concurrency=False,
force_update=False,
)

snet_sdk = sdk.SnetSDK(config) # the 'SnetSDK' instance
initialized_services = [] # the list of initialized service clients
Expand All @@ -319,42 +367,72 @@
"""
commands = {
"main": {
"organizations": (list_organizations, "print a list of organization ids from Registry"),
"services": (list_services_for_org, "print a list of service ids for an organization from Registry"),
"balance": (balance, "print the account balance and the escrow balance"),
"organizations": (
list_organizations,
"print a list of organization ids from Registry",
),
"services": (
list_services_for_org,
"print a list of service ids for an organization from Registry",
),
"balance": (
balance,
"print the account balance and the escrow balance",
),
"deposit": (deposit, "deposit FET tokens into MPE"),
"block": (block_number, "print the current block number"),
"service": (lambda: None, "go to the services menu"),
"channel": (lambda: None, "go to the channels menu"),
"help": (commands_help, "print a list of available commands in the main menu"),
"exit": (lambda: exit(0), "exit the application")
"help": (
commands_help,
"print a list of available commands in the main menu",
),
"exit": (lambda: exit(0), "exit the application"),

Check failure on line 390 in examples/console_app.py

View check run for this annotation

snet-sonarqube-app / SonarQube Code Analysis

examples/console_app.py#L390

Define a constant instead of duplicating this literal "exit the application" 3 times.
},

"service": {
"add": (create_service_client,
"create a new service client. If it the first time, the new service becomes active"),
"add": (
create_service_client,
"create a new service client. If it the first time, the new service becomes active",
),
"use": (switch_service, "switch the active service"),
"call": (call, "call the active service method"),
"info": (print_service_info, "output services, methods and messages in a service"),
"list": (list_initialized_services, "print a list of initialized services"),
"help": (commands_help, "print a list of available commands in the services menu"),
"info": (
print_service_info,
"output services, methods and messages in a service",
),
"list": (
list_initialized_services,
"print a list of initialized services",
),
"help": (
commands_help,
"print a list of available commands in the services menu",
),
"back": (lambda: None, "return to the main menu"),
"exit": (lambda: exit(0), "exit the application")
"exit": (lambda: exit(0), "exit the application"),
},

"channel": {
"update": (update_channels, "update a list of initialized payment channels"),
"update": (
update_channels,
"update a list of initialized payment channels",
),
"list": (list_channels, "print a list of initialized payment channels"),
"open": (open_channel, "open a new payment channel"),
"add-funds": (add_funds, "add funds to a channel"),
"extend-expiration": (extend_expiration, "extend expiration of a channel"),
"help": (commands_help, "print a list of available commands in the channels menu"),
"extend-expiration": (
extend_expiration,
"extend expiration of a channel",
),
"help": (
commands_help,
"print a list of available commands in the channels menu",
),
"back": (lambda: None, "return to the main menu"),
"exit": (lambda: exit(0), "exit the application")
}
"exit": (lambda: exit(0), "exit the application"),
},
}

active_commands: dict = commands["main"] # the list of available commands in the active menu
active_commands: dict = commands["main"] # the list of available commands in the active menu


def main():
Expand Down
Loading
Loading