[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[libeufin] branch master updated: don't clutter source tree
|
From: |
gnunet |
|
Subject: |
[libeufin] branch master updated: don't clutter source tree |
|
Date: |
Fri, 22 Jan 2021 17:15:55 +0100 |
This is an automated email from the git hooks/post-receive script.
dold pushed a commit to branch master
in repository libeufin.
The following commit(s) were added to refs/heads/master by this push:
new d6f9f6c don't clutter source tree
d6f9f6c is described below
commit d6f9f6c931522fe57e7a7e0029235c8f1b4d454e
Author: Florian Dold <florian@dold.me>
AuthorDate: Fri Jan 22 17:15:45 2021 +0100
don't clutter source tree
---
build.gradle | 10 +-
cli/bin/libeufin-cli.template | 1048 -----------------------------------------
2 files changed, 5 insertions(+), 1053 deletions(-)
diff --git a/build.gradle b/build.gradle
index 86153d6..6ded440 100644
--- a/build.gradle
+++ b/build.gradle
@@ -31,9 +31,8 @@ task versionFile() {
}
task replaceVersionCli(type: Copy) {
- from file("cli/bin/libeufin-cli.template")
- into file("cli/bin/")
- rename("libeufin-cli.template", "libeufin-cli")
+ from file("cli/bin/libeufin-cli")
+ into file("$project.buildDir/generated/python")
filter(ReplaceTokens, tokens: [version: getRootProject().version])
}
@@ -63,8 +62,9 @@ task dist(type: Zip) {
include("**/libeufin-sandbox")
include("**/*.jar")
}
- from ("cli") {
- include("**/libeufin-cli")
+ from("$project.buildDir/generated/python") {
+ include("libeufin-cli")
+ rename { "bin/libeufin-cli" }
}
into(topDir)
}
diff --git a/cli/bin/libeufin-cli.template b/cli/bin/libeufin-cli.template
deleted file mode 100755
index 724aa20..0000000
--- a/cli/bin/libeufin-cli.template
+++ /dev/null
@@ -1,1048 +0,0 @@
-#!/usr/bin/env python3
-
-import os
-import sys
-import click
-import json
-import hashlib
-import errno
-from datetime import datetime
-from requests import post, get, auth, delete
-from urllib.parse import urljoin
-from getpass import getpass
-
-
-def tell_user(resp, withsuccess=False):
- if resp.status_code == 200 and not withsuccess:
- return
- print(resp.content.decode("utf-8"))
-
-
-# FIXME: deprecate this in favor of NexusContext
-def fetch_env():
- if "--help" in sys.argv:
- return []
- try:
- nexus_base_url = os.environ["LIBEUFIN_NEXUS_URL"]
- nexus_username = os.environ["LIBEUFIN_NEXUS_USERNAME"]
- nexus_password = os.environ["LIBEUFIN_NEXUS_PASSWORD"]
- except KeyError:
- print(
- "Please ensure that NEXUS_BASE_URL,"
- " NEXUS_USERNAME, NEXUS_PASSWORD exist"
- " in the environment"
- )
- sys.exit(1)
- return nexus_base_url, nexus_username, nexus_password
-
-
-# FIXME: deprecate this in favor of NexusContext
-class NexusAccess:
- def __init__(self, nexus_base_url=None, username=None, password=None):
- self.nexus_base_url = nexus_base_url
- self.username = username
- self.password = password
-
-
-@click.group(help="General utility to invoke HTTP REST services offered by
Nexus.")
-@click.version_option(version="@version@")
-def cli():
- pass
-
-
-@cli.group()
-@click.pass_context
-def facades(ctx):
- ctx.obj = NexusAccess(*fetch_env())
-
-
-@cli.group()
-@click.pass_context
-def connections(ctx):
- ctx.obj = NexusAccess(*fetch_env())
-
-
-@cli.group()
-@click.pass_context
-def users(ctx):
- ctx.obj = NexusContext()
-
-@cli.group()
-@click.pass_context
-def permissions(ctx):
- ctx.obj = NexusContext()
-
-@users.command("list", help="List users")
-@click.pass_obj
-def list_users(obj):
- url = urljoin(obj.nexus_base_url, f"/users")
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.nexus_username,
obj.nexus_password))
- except Exception as e:
- print(e)
- print("Could not reach nexus at " + url)
- exit(1)
-
- print(resp.content.decode("utf-8"))
-
-
-@users.command("create", help="Create a new user")
-@click.argument("username")
-@click.option(
- "--password",
- help="Provide password instead of prompting interactively.",
- prompt=True,
- hide_input=True,
- confirmation_prompt=True,
-)
-@click.pass_obj
-def create_user(obj, username, password):
- url = urljoin(obj.nexus_base_url, f"/users")
- try:
- body = dict(
- username=username,
- password=password,
- )
- resp = post(
- url,
- json=body,
- auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password),
- )
- except Exception as e:
- print(e)
- print("Could not reach nexus at " + url)
- exit(1)
-
- print(resp.content.decode("utf-8"))
-
-
-@permissions.command("list", help="Show permissions")
-@click.pass_obj
-def list_permission(obj):
- url = urljoin(obj.nexus_base_url, f"/permissions")
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.nexus_username,
obj.nexus_password))
- except Exception as e:
- print(e)
- print("Could not reach nexus at " + url)
- exit(1)
-
- print(resp.content.decode("utf-8"))
-
-@permissions.command("grant", help="Grant permission to a subject")
-@click.pass_obj
-@click.argument("subject-type")
-@click.argument("subject-id")
-@click.argument("resource-type")
-@click.argument("resource-id")
-@click.argument("permission-name")
-def grant_permission(obj, subject_type, subject_id, resource_type,
resource_id, permission_name):
- url = urljoin(obj.nexus_base_url, f"/permissions")
- try:
- permission = dict(
- subjectType=subject_type,
- subjectId=subject_id,
- resourceType=resource_type,
- resourceId=resource_id,
- permissionName=permission_name,
- )
- body = dict(
- permission=permission,
- action="grant",
- )
- resp = post(url, json=body,
auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password))
- except Exception as e:
- print(e)
- print("Could not reach nexus at " + url)
- exit(1)
-
- print(resp.content.decode("utf-8"))
-
-@permissions.command("revoke", help="Revoke permission from a subject")
-@click.pass_obj
-@click.argument("subject-type")
-@click.argument("subject-id")
-@click.argument("resource-type")
-@click.argument("resource-id")
-@click.argument("permission-name")
-def grant_permission(obj, subject_type, subject_id, resource_type,
resource_id, permission_name):
- url = urljoin(obj.nexus_base_url, f"/permissions")
- try:
- permission = dict(
- subjectType=subject_type,
- subjectId=subject_id,
- resourceType=resource_type,
- resourceId=resource_id,
- permissionName=permission_name,
- )
- body = dict(
- permission=permission,
- action="revoke",
- )
- resp = post(url, json=body,
auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password))
- except Exception as e:
- print(e)
- print("Could not reach nexus at " + url)
- exit(1)
-
- print(resp.content.decode("utf-8"))
-
-
-@cli.group()
-@click.pass_context
-def accounts(ctx):
- ctx.obj = NexusAccess(*fetch_env())
-
-
-class SandboxContext:
- def __init__(self):
- self.sandbox_base_url = None
-
- def require_sandbox_base_url(self):
- if self.sandbox_base_url:
- return self.sandbox_base_url
- sandbox_base_url = os.environ.get("LIBEUFIN_SANDBOX_URL")
- if not sandbox_base_url:
- raise click.UsageError(
- "sandbox URL must be given as an argument or in
LIBEUFIN_SANDBOX_URL"
- )
- return sandbox_base_url
-
-
-class NexusContext:
- def __init__(self):
- self._nexus_base_url = None
- self._nexus_password = None
- self._nexus_username = None
-
- @property
- def nexus_base_url(self):
- if self._nexus_base_url:
- return self._nexus_base_url
- val = os.environ.get("LIBEUFIN_NEXUS_URL")
- if not val:
- raise click.UsageError(
- "nexus URL must be given as an argument or in
LIBEUFIN_NEXUS_URL"
- )
- self._nexus_base_url = val
- return val
-
- @property
- def nexus_username(self):
- if self._nexus_username:
- return self._nexus_username
- val = os.environ.get("LIBEUFIN_NEXUS_USERNAME")
- if not val:
- raise click.UsageError(
- "nexus username must be given as an argument or in
LIBEUFIN_NEXUS_USERNAME"
- )
- self._nexus_username = val
- return val
-
- @property
- def nexus_password(self):
- if self._nexus_password:
- return self._nexus_password
- val = os.environ.get("LIBEUFIN_NEXUS_PASSWORD")
- if not val:
- raise click.UsageError(
- "nexus password must be given as an argument or in
LIBEUFIN_NEXUS_PASSWORD"
- )
- self._nexus_password = val
- return val
-
-
-@cli.group()
-@click.option("--sandbox-url", help="URL for the sandbox", required=False)
-@click.pass_context
-def sandbox(ctx, sandbox_url):
- ctx.obj = SandboxContext()
- ctx.obj.sandbox_base_url = sandbox_url
-
-
-@connections.command(help="Get key letter (typically PDF).")
-@click.argument("connection-name")
-@click.argument("output_file")
-@click.pass_obj
-def get_key_letter(obj, connection_name, output_file):
- url = urljoin(obj.nexus_base_url,
f"/bank-connections/{connection_name}/keyletter")
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password))
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- if resp.status_code != 200:
- print(resp.content.decode("utf-8"))
- sys.exit(1)
-
- output = open(output_file, "wb")
- output.write(resp.content)
- output.close()
-
-
-@connections.command(help="export backup")
-@click.option("--passphrase", help="Passphrase for locking the backup",
required=True)
-@click.option("--output-file", help="Where to store the backup", required=True)
-@click.argument("connection-name")
-@click.pass_obj
-def export_backup(obj, connection_name, passphrase, output_file):
- url = urljoin(
- obj.nexus_base_url,
"/bank-connections/{}/export-backup".format(connection_name)
- )
- try:
- resp = post(
- url,
- json=dict(passphrase=passphrase),
- auth=auth.HTTPBasicAuth(obj.username, obj.password),
- )
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- output = open(output_file, "w+")
- output.write(resp.content.decode("utf-8"))
- output.close()
-
- print("Backup stored in {}".format(output_file))
-
-
-@connections.command(help="delete bank connection")
-@click.argument("connection-name")
-@click.pass_obj
-def delete_connection(obj, connection_name):
-
- url = urljoin(obj.nexus_base_url, "/bank-connections/delete-connection")
- try:
- resp = post(
- url,
- json=dict(bankConnectionId=connection_name),
- auth=auth.HTTPBasicAuth(obj.username, obj.password),
- )
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp)
-
-
-@connections.command(help="restore backup")
-@click.option("--backup-file", help="Back file", required=True)
-@click.option("--passphrase", help="Passphrase for locking the backup",
required=True)
-@click.argument("connection-name")
-@click.pass_obj
-def restore_backup(obj, backup_file, passphrase, connection_name):
- url = urljoin(obj.nexus_base_url, "/bank-connections")
- try:
- backup = open(backup_file, "r")
- except Exception:
- print("Could not open the backup at {}".format(backup_file))
- exit(1)
- backup_json = json.loads(backup.read())
- backup.close()
-
- try:
- resp = post(
- url,
- json=dict(
- name=connection_name,
- data=backup_json,
- passphrase=passphrase,
- source="backup",
- ),
- auth=auth.HTTPBasicAuth(obj.username, obj.password),
- )
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp)
-
-
-@connections.command(help="make new EBICS bank connection")
-@click.option("--ebics-url", help="EBICS URL", required=True)
-@click.option("--host-id", help="Host ID", required=True)
-@click.option("--partner-id", help="Partner ID", required=True)
-@click.option("--ebics-user-id", help="Ebics user ID", required=True)
-@click.argument("connection-name")
-@click.pass_obj
-def new_ebics_connection(
- obj, connection_name, ebics_url, host_id, partner_id, ebics_user_id
-):
- url = urljoin(obj.nexus_base_url, "/bank-connections")
- body = dict(
- name=connection_name,
- source="new",
- type="ebics",
- data=dict(
- ebicsURL=ebics_url,
- hostID=host_id,
- partnerID=partner_id,
- userID=ebics_user_id,
- ),
- )
- try:
- resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.username,
obj.password))
- except Exception:
- print(f"Could not reach nexus at {url}")
- exit(1)
-
- tell_user(resp)
-
-
-@connections.command(help="Initialize the bank connection.")
-@click.argument("connection-name")
-@click.pass_obj
-def connect(obj, connection_name):
- url = urljoin(obj.nexus_base_url,
f"/bank-connections/{connection_name}/connect")
- try:
- resp = post(
- url, json=dict(), auth=auth.HTTPBasicAuth(obj.username,
obj.password)
- )
- except Exception:
- print(f"Could not reach nexus at {url}")
- exit(1)
- tell_user(resp)
-
-
-@connections.command(help="Import one bank account, chosen from the downloaded
ones.")
-@click.option(
- "--offered-account-id", help="Name of the account to import", required=True
-)
-@click.option(
- "--nexus-bank-account-id",
- help="Name to give to the imported account",
- required=True,
-)
-@click.argument("connection-name")
-@click.pass_obj
-def import_bank_account(
- obj, connection_name, offered_account_id, nexus_bank_account_id
-):
- url = urljoin(
- obj.nexus_base_url,
- "/bank-connections/{}/import-account".format(connection_name),
- )
- try:
- resp = post(
- url,
- json=dict(
- offeredAccountId=offered_account_id,
- nexusBankAccountId=nexus_bank_account_id,
- ),
- auth=auth.HTTPBasicAuth(obj.username, obj.password),
- )
- except Exception as e:
- print(f"Could not reach nexus at {url}: {e}")
- exit(1)
-
- tell_user(resp)
-
-
-@connections.command(help="Update list of bank accounts available through this
connection.")
-@click.argument("connection-name")
-@click.pass_obj
-def download_bank_accounts(obj, connection_name):
- url = urljoin(
- obj.nexus_base_url,
- "/bank-connections/{}/fetch-accounts".format(connection_name),
- )
- try:
- resp = post(
- url, json=dict(), auth=auth.HTTPBasicAuth(obj.username,
obj.password)
- )
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp)
-
-
-@connections.command(help="List the connections.")
-@click.pass_obj
-def list_connections(obj):
- url = urljoin(obj.nexus_base_url, "/bank-connections")
- try:
- resp = get(
- url, json=dict(), auth=auth.HTTPBasicAuth(obj.username,
obj.password)
- )
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@connections.command(help="Show the status of a bank connection.")
-@click.argument("connection-name")
-@click.pass_obj
-def show_connection(obj, connection_name):
- url = urljoin(obj.nexus_base_url, f"/bank-connections/{connection_name}")
- try:
- resp = get(
- url, json=dict(), auth=auth.HTTPBasicAuth(obj.username,
obj.password)
- )
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@connections.command(help="list bank accounts hosted at one connection")
-@click.argument("connection-name")
-@click.pass_obj
-def list_offered_bank_accounts(obj, connection_name):
- url = urljoin(
- obj.nexus_base_url,
"/bank-connections/{}/accounts".format(connection_name)
- )
- try:
- resp = get(
- url, json=dict(), auth=auth.HTTPBasicAuth(obj.username,
obj.password)
- )
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@accounts.command(help="Schedules a new task")
-@click.argument("account-name")
-@click.option("--task-name", help="Name of the task", required=True)
-@click.option("--task-cronspec", help="Cronspec string", required=True)
-@click.option(
- "--task-type",
- help="'fetch' (downloads transactions histories) or 'submit' (uploads
payments instructions)",
- required=True,
-)
-@click.option(
- "--task-param-range-type",
- help="Only needed for 'fetch'. (FIXME: link to documentation here!)",
- required=False,
-)
-@click.option(
- "--task-param-level",
- help="Only needed for 'fetch'. (FIXME: link to documentation here!)",
- required=False,
-)
-@click.pass_obj
-def task_schedule(
- obj,
- account_name,
- task_name,
- task_cronspec,
- task_type,
- task_param_range_type,
- task_param_level,
-):
-
- url = urljoin(obj.nexus_base_url,
"/bank-accounts/{}/schedule".format(account_name))
- body = dict(name=task_name, cronspec=task_cronspec, type=task_type)
- if task_type == "fetch" and not (task_param_range_type or
task_param_level):
- print("'fetch' type requires --task-param-range-type and
--task-param-level")
- return
-
- body.update(
- dict(params=dict(rangeType=task_param_range_type,
level=task_param_level))
- )
- try:
- resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.username,
obj.password))
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp)
-
-
-@accounts.command(help="Shows the status of one task")
-@click.argument("account-name")
-@click.option("--task-name", help="Name of the task", required=True)
-@click.pass_obj
-def task_status(obj, account_name, task_name):
- url = urljoin(
- obj.nexus_base_url,
- "/bank-accounts/{}/schedule/{}".format(account_name, task_name),
- )
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password))
- except Exception:
- print("Could not reach nexus " + url)
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@accounts.command(help="Deletes one task")
-@click.argument("account-name")
-@click.option("--task-name", help="Name of the task", required=True)
-@click.pass_obj
-def task_delete(obj, account_name, task_name):
- url = urljoin(
- obj.nexus_base_url,
- "/bank-accounts/{}/schedule/{}".format(account_name, task_name),
- )
- try:
- resp = delete(url, auth=auth.HTTPBasicAuth(obj.username, obj.password))
- except Exception:
- print("Could not reach nexus " + url)
- exit(1)
-
- tell_user(resp)
-
-
-@accounts.command(help="Shows all the active tasks")
-@click.argument("account-name")
-@click.pass_obj
-def tasks_show(obj, account_name):
- url = urljoin(obj.nexus_base_url,
"/bank-accounts/{}/schedule".format(account_name))
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password))
- except Exception:
- print("Could not reach nexus " + url)
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@accounts.command(help="show accounts belonging to calling user")
-@click.pass_obj
-def show(obj):
- url = urljoin(obj.nexus_base_url, "/bank-accounts")
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password))
- except Exception as e:
- print(f"Could not reach nexus at {url}, error: {e}")
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@accounts.command(help="prepare payment debiting 'account-name'")
-@click.option(
- "--creditor-iban", help="IBAN that will receive the payment", required=True
-)
-@click.option(
- "--creditor-bic", help="BIC that will receive the payment", required=False
-)
-@click.option(
- "--creditor-name", help="Legal name that will receive the payment",
required=True
-)
-@click.option(
- "--payment-amount", help="Amount to be paid (<currency>:X.Y)",
required=True
-)
-@click.option("--payment-subject", help="Subject of this payment",
required=True)
-@click.argument("account-name")
-@click.pass_obj
-def prepare_payment(
- obj,
- account_name,
- creditor_iban,
- creditor_bic,
- creditor_name,
- payment_amount,
- payment_subject,
-):
- url = urljoin(
- obj.nexus_base_url,
"/bank-accounts/{}/payment-initiations".format(account_name)
- )
- body = dict(
- iban=creditor_iban,
- bic=creditor_bic,
- name=creditor_name,
- subject=payment_subject,
- amount=payment_amount,
- )
-
- try:
- resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.username,
obj.password))
- except Exception:
- print("Could not reach nexus at " + url)
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@accounts.command(help="submit a prepared payment")
-@click.option("--payment-uuid", help="payment unique identifier",
required=True)
-@click.argument("account-name")
-@click.pass_obj
-def submit_payment(obj, account_name, payment_uuid):
- url = urljoin(
- obj.nexus_base_url,
- "/bank-accounts/{}/payment-initiations/{}/submit".format(
- account_name, payment_uuid
- ),
- )
- try:
- resp = post(
- url, json=dict(), auth=auth.HTTPBasicAuth(obj.username,
obj.password)
- )
- except Exception:
- print("Could not reach nexus at" + url)
- exit(1)
-
- tell_user(resp)
-
-
-@accounts.command(help="fetch transactions from the bank")
-@click.option(
- "--range-type",
- default="all",
- help="Admitted values: all, latest, previous-days, since-last",
-)
-@click.option("--level", default="all", help="Admitted values: report,
statement, all")
-@click.argument("account-name")
-@click.pass_obj
-def fetch_transactions(obj, account_name, range_type, level):
- url = urljoin(
- obj.nexus_base_url,
"/bank-accounts/{}/fetch-transactions".format(account_name)
- )
- try:
- resp = post(
- url,
- json=dict(rangeType=range_type, level=level),
- auth=auth.HTTPBasicAuth(obj.username, obj.password),
- )
- except Exception:
- print("Could not reach nexus " + url)
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@accounts.command(help="get transactions from the simplified nexus JSON API")
-@click.option(
- "--compact/--no-compact",
- help="Tells only amount/subject for each payment",
- required=False,
- default=False,
-)
-@click.argument("account-name")
-@click.pass_obj
-def transactions(obj, compact, account_name):
- url = urljoin(
- obj.nexus_base_url,
"/bank-accounts/{}/transactions".format(account_name)
- )
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password))
- except Exception:
- print("Could not reach nexus " + url)
- exit(1)
-
- if compact and resp.status_code == 200:
- for payment in resp.json()["transactions"]:
- for entry in payment["batches"]:
- for expected_singleton in entry["batchTransactions"]:
- print(
- "{}, {}".format(
- expected_singleton["details"][
- "unstructuredRemittanceInformation"
- ],
- expected_singleton["amount"],
- )
- )
- return
-
- tell_user(resp, withsuccess=True)
-
-
-@facades.command(help="List active facades in the Nexus")
-@click.argument("connection-name")
-@click.pass_obj
-def list_facades(obj, connection_name):
- url = urljoin(obj.nexus_base_url, "/facades")
- try:
- resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password))
- except Exception as e:
- print(f"Could not reach nexus (at {obj.nexus_base_url}): {e}")
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@facades.command(help="create a new (Taler) facade")
-@click.option("--facade-name", help="Name of the facade", required=True)
-@click.option("--currency", help="Facade's currency", required=True)
-@click.argument("connection-name")
-@click.argument("account-name")
-@click.pass_obj
-def new_facade(obj, facade_name, connection_name, account_name, currency):
- url = urljoin(obj.nexus_base_url, "/facades")
- try:
- resp = post(
- url,
- auth=auth.HTTPBasicAuth(obj.username, obj.password),
- json=dict(
- name=facade_name,
- type="taler-wire-gateway",
- config=dict(
- currency=currency,
- bankAccount=account_name,
- bankConnection=connection_name,
- reserveTransferLevel="UNUSED",
- intervalIncremental="UNUSED",
- ),
- ),
- )
- except Exception as e:
- print(f"Could not reach nexus (at {obj.nexus_base_url}): {e}")
- exit(1)
-
- tell_user(resp)
-
-
-@sandbox.group("ebicshost", help="manage EBICS hosts")
-@click.pass_context
-def sandbox_ebicshost(ctx):
- pass
-
-
-@sandbox.command("check", help="check sandbox status")
-@click.pass_obj
-def check_sandbox_status(obj):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, "/config")
- try:
- resp = get(url)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@sandbox_ebicshost.command("create", help="Create an EBICS host")
-@click.option("--host-id", help="EBICS host ID", required=True, prompt=True)
-@click.pass_obj
-def make_ebics_host(obj, host_id):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, "/admin/ebics/hosts")
- try:
- resp = post(url, json=dict(hostID=host_id, ebicsVersion="2.5"))
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp)
-
-
-@sandbox_ebicshost.command("list", help="List EBICS hosts.")
-@click.pass_obj
-def list_ebics_host(obj):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, "/admin/ebics/hosts")
- try:
- resp = get(url)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@sandbox.group("ebicssubscriber", help="manage EBICS subscribers")
-@click.pass_context
-def sandbox_ebicssubscriber(ctx):
- pass
-
-
-@sandbox_ebicssubscriber.command("create", help="Create an EBICS subscriber.")
-@click.option("--host-id", help="Ebics host ID", required=True, prompt=True)
-@click.option("--partner-id", help="Ebics partner ID", required=True,
prompt=True)
-@click.option("--user-id", help="Ebics user ID", required=True, prompt=True)
-@click.pass_obj
-def create_ebics_subscriber(obj, host_id, partner_id, user_id):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, "/admin/ebics/subscribers")
- try:
- resp = post(
- url, json=dict(hostID=host_id, partnerID=partner_id,
userID=user_id)
- )
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp)
-
-
-@sandbox_ebicssubscriber.command("list", help="List EBICS subscribers.")
-@click.pass_obj
-def list_ebics_subscriber(obj):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, "/admin/ebics/subscribers")
- try:
- resp = get(url)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@sandbox.group("ebicsbankaccount", help="manage EBICS bank accounts")
-@click.pass_context
-def sandbox_ebicsbankaccount(ctx):
- pass
-
-
-@sandbox_ebicsbankaccount.command(
- "create", help="Create a bank account for a EBICS subscriber."
-)
-@click.option("--currency", help="currency", prompt=True)
-@click.option("--iban", help="IBAN", required=True)
-@click.option("--bic", help="BIC", required=True)
-@click.option("--person-name", help="bank account owner name", required=True)
-@click.option("--account-name", help="label of this bank account",
required=True)
-@click.option("--ebics-user-id", help="user ID of the Ebics subscriber",
required=True)
-@click.option("--ebics-host-id", help="host ID of the Ebics subscriber",
required=True)
-@click.option(
- "--ebics-partner-id", help="partner ID of the Ebics subscriber",
required=True
-)
-@click.pass_obj
-def associate_bank_account(
- obj,
- currency,
- iban,
- bic,
- person_name,
- account_name,
- ebics_user_id,
- ebics_host_id,
- ebics_partner_id,
-):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, "/admin/ebics/bank-accounts")
- body = dict(
- currency=currency,
- subscriber=dict(
- userID=ebics_user_id, partnerID=ebics_partner_id,
hostID=ebics_host_id
- ),
- iban=iban,
- bic=bic,
- name=person_name,
- label=account_name,
- )
-
- try:
- resp = post(url, json=body)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp)
-
-
-@sandbox.group("bankaccount", help="manage bank accounts")
-@click.pass_context
-def sandbox_bankaccount(ctx):
- pass
-
-
-@sandbox_bankaccount.command("list", help="List accounts")
-@click.pass_obj
-def bankaccount_list(obj):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, f"/admin/bank-accounts")
- try:
- resp = get(url)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@sandbox_bankaccount.command("transactions", help="List transactions")
-@click.argument("account-label")
-@click.pass_obj
-def transactions_list(obj, account_label):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(
- sandbox_base_url, f"/admin/bank-accounts/{account_label}/transactions"
- )
- try:
- resp = get(url)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp, withsuccess=True)
-
-
-@sandbox_bankaccount.command("generate-transactions", help="Generate test
transactions")
-@click.argument("account-label")
-@click.pass_obj
-def bankaccount_generate_transactions(obj, account_label):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(
- sandbox_base_url,
f"/admin/bank-accounts/{account_label}/generate-transactions"
- )
- try:
- resp = post(url)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp)
-
-
-@sandbox_bankaccount.command(help="Book a payment in the sandbox")
-@click.option("--creditor-iban", help="IBAN receiving the payment",
prompt=True)
-@click.option("--creditor-bic", help="BIC receiving the payment", prompt=True)
-@click.option(
- "--creditor-name",
- help="Name of the person who is receiving the payment",
- prompt=True,
-)
-@click.option("--debtor-iban", help="IBAN sending the payment", prompt=True)
-@click.option("--debtor-bic", help="BIC sending the payment", prompt=True)
-@click.option(
- "--debtor-name", help="name of the person who is sending the payment",
prompt=True
-)
-@click.option("--amount", help="amount, no currency", prompt=True)
-@click.option("--currency", help="currency", prompt=True)
-@click.option("--subject", help="payment subject", prompt=True)
-@click.option(
- "--direction",
- help="direction respect to the bank account hosted at Sandbox: allows
DBIT/CRDT values.",
- prompt=True,
-)
-@click.pass_obj
-def book_payment(
- obj,
- creditor_iban,
- creditor_bic,
- creditor_name,
- debtor_iban,
- debtor_bic,
- debtor_name,
- amount,
- currency,
- subject,
- direction,
-):
- sandbox_base_url = obj.require_sandbox_base_url()
- url = urljoin(sandbox_base_url, "/admin/payments")
- body = dict(
- creditorIban=creditor_iban,
- creditorBic=creditor_bic,
- creditorName=creditor_name,
- debitorIban=debtor_iban,
- debitorBic=debtor_bic,
- debitorName=debtor_name,
- amount=amount,
- currency=currency,
- subject=subject,
- direction=direction,
- )
- try:
- resp = post(url, json=body)
- except Exception:
- print("Could not reach sandbox")
- exit(1)
-
- tell_user(resp)
-
-
-cli(obj={})
--
To stop receiving notification emails like this one, please contact
gnunet@gnunet.org.
| [Prev in Thread] |
Current Thread |
[Next in Thread] |
- [libeufin] branch master updated: don't clutter source tree,
gnunet <=