diff --git a/frontend/admincli/pol.py b/frontend/admincli/pol.py
index 4a51c0d8822ccef4c5677ed763d7b68855ee0c3f..7007c3edb9c36222cd80175fade4803bf4b2db0a 100755
--- a/frontend/admincli/pol.py
+++ b/frontend/admincli/pol.py
@@ -4,7 +4,6 @@ from time import time
import typer
from typing import List, Optional
from types import SimpleNamespace
-#from rich import print
from rich.table import Table
from rich.console import Console
from rich.tree import Tree
@@ -26,13 +25,14 @@ from rego import ast, walk
import opa
import search_cli
-__version__ = '0.0.5'
+__version__ = '0.0.6'
__app_name__ = 'OSDU Policy Service AdminCLI'
cli = typer.Typer(rich_markup_mode="rich", help=__app_name__)
console = Console()
error_console = Console(stderr=True, style="bold red")
+
# setup simple output class
class OutputType(str, Enum):
simple = "simple"
@@ -40,26 +40,28 @@ class OutputType(str, Enum):
tree = "tree"
excel = "excel"
+
def _version_callback(value: bool) -> None:
if value:
typer.echo(f"{__app_name__} v{__version__}")
raise typer.Exit()
+
def headers(ctx: typer.Context, content_type_json=False):
"""
build headers for OSDU request
"""
if content_type_json:
- headers={'Authorization': 'Bearer ' + ctx.obj.token,
- 'data-partition-id': ctx.obj.data_partition,
- 'accept': 'application/json',
- 'content-type': 'application/json'
- }
+ headers = {'Authorization': 'Bearer ' + ctx.obj.token,
+ 'data-partition-id': ctx.obj.data_partition,
+ 'accept': 'application/json',
+ 'content-type': 'application/json'
+ }
else:
- headers={'Authorization': 'Bearer ' + ctx.obj.token,
- 'data-partition-id': ctx.obj.data_partition,
- 'accept': 'application/json',
- }
+ headers = {'Authorization': 'Bearer ' + ctx.obj.token,
+ 'data-partition-id': ctx.obj.data_partition,
+ 'accept': 'application/json',
+ }
if ctx.obj.uuid:
headers['Correlation-id'] = ctx.obj.uuid
if ctx.obj.x_user_id:
@@ -70,13 +72,15 @@ def headers(ctx: typer.Context, content_type_json=False):
console.print(f"headers: {headers}")
return headers
+
def request_groups(ctx: typer.Context):
"""
return list of groups from entitlements service
"""
try:
- response = requests.get(ctx.obj.entitlements_url + "/groups", headers = headers(ctx))
- except requests.exceptions.RequestException as e:
+ response = requests.get(
+ ctx.obj.entitlements_url + "/groups", headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if not response.ok:
@@ -84,20 +88,22 @@ def request_groups(ctx: typer.Context):
message = json.loads(response.text)["message"]
error_console.print(f"Error: {message}")
else:
- error_console.print(f"Error: An error occurred when talking to {ctx.obj.entitlements_url} {response.status_code}")
+ error_console.print(
+ f"Error: An error occurred when talking to {ctx.obj.entitlements_url} {response.status_code}")
return ""
jdata = response.json()
if "groups" in jdata:
return jdata['groups']
+
def get_policies(ctx: typer.Context):
"""
Handle request to /policies
"""
try:
r = requests.get(ctx.obj.url + "/policies",
- timeout=10,
- headers=headers(ctx))
+ timeout=10,
+ headers=headers(ctx))
except requests.exceptions.HTTPError:
error_console.print(f"Error: endpoint {ctx.obj.url}: HTTPError")
return
@@ -112,13 +118,14 @@ def get_policies(ctx: typer.Context):
return
if "result" in r.text and r.ok:
- return(r.json()["result"])
+ return (r.json()["result"])
else:
if "detail" in r.text:
error_console.print(f"Error: {r.json()['detail']}")
else:
error_console.print(f"Error: {r.text}")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def delete_partition_policy(ctx: typer.Context, policy_id: str):
"""
@@ -126,8 +133,8 @@ def delete_partition_policy(ctx: typer.Context, policy_id: str):
"""
try:
r = requests.delete(ctx.obj.url + "/policies/osdu/partition/" + ctx.obj.data_partition + '/' + policy_id,
- headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if ctx.obj.debug:
@@ -140,7 +147,8 @@ def delete_partition_policy(ctx: typer.Context, policy_id: str):
error_console.print(f"Error: {r.json()['detail']}")
else:
error_console.print(f"Error: {r.text}")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def add_partition_policy(ctx: typer.Context, policy_id: str, files: dict):
"""
@@ -148,9 +156,9 @@ def add_partition_policy(ctx: typer.Context, policy_id: str, files: dict):
"""
try:
r = requests.put(ctx.obj.url + "/policies/osdu/partition/" + ctx.obj.data_partition + '/' + policy_id,
- files=files,
- headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ files=files,
+ headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
@@ -160,7 +168,8 @@ def add_partition_policy(ctx: typer.Context, policy_id: str, files: dict):
error_console.print(f"Error: {r.json()['detail']}")
else:
error_console.print(f"Error: {r.text}")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def get_partition_policy(ctx: typer.Context, policy_id: str, quiet=False):
"""
@@ -168,19 +177,20 @@ def get_partition_policy(ctx: typer.Context, policy_id: str, quiet=False):
"""
try:
r = requests.get(ctx.obj.url + "/policies/osdu/partition/" + ctx.obj.data_partition + '/' + policy_id,
- headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
if "result" in r.text:
- return(r.json()["result"])
+ return (r.json()["result"])
else:
if not quiet and "detail" in r.text:
error_console.print(f"Error: {r.json()['detail']}")
else:
error_console.print(f"Error: {r.text}")
- #raise typer.Exit(1) # non-zero exit status
+ # raise typer.Exit(1) # non-zero exit status
+
def get_instance_policy(ctx: typer.Context, policy_id: str, quiet=False):
"""
@@ -188,19 +198,20 @@ def get_instance_policy(ctx: typer.Context, policy_id: str, quiet=False):
"""
try:
r = requests.get(ctx.obj.url + "/policies/osdu/instance/" + policy_id,
- headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
if "result" in r.text:
- return(r.json()["result"])
+ return (r.json()["result"])
else:
if not quiet and "detail" in r.text:
error_console.print(f"Error: {r.json()['detail']}")
else:
error_console.print(f"Error: {r.text}")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def evaluations_query(ctx: typer.Context, policy_id: str, files: dict):
"""
@@ -209,31 +220,32 @@ def evaluations_query(ctx: typer.Context, policy_id: str, files: dict):
params = {
'policy_id': policy_id,
'include_auth': True
- }
+ }
try:
r = requests.post(ctx.obj.url + "/evaluations/query",
- params=params,
- files=files,
- headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ params=params,
+ files=files,
+ headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
if "result" in r.text:
- return(r.json()["result"])
+ return (r.json()["result"])
else:
if "detail" in r.text:
error_console.print(f"Error: {r.json()['detail']}")
else:
error_console.print(f"Error: {r.text}")
- raise typer.Exit(r.status_code) # non-zero exit status
+ raise typer.Exit(r.status_code) # non-zero exit status
+
@cli.command(rich_help_panel="Utils")
def groups(ctx: typer.Context,
- all: bool = False,
- domain: bool = False,
- domain_suffix: bool = False):
+ all: bool = False,
+ domain: bool = False,
+ domain_suffix: bool = False):
"""
Show groups of current auth context.
"""
@@ -241,11 +253,11 @@ def groups(ctx: typer.Context,
if all:
console.print(retgroups)
elif domain:
- domain=retgroups[0]["email"].split('@')[1]
+ domain = retgroups[0]["email"].split('@')[1]
domain_ending = domain.removeprefix(ctx.obj.data_partition + ".")
console.print(f"fulldomain: {domain}\ndomain: {domain_ending}")
elif domain_suffix:
- domain=retgroups[0]["email"].split('@')[1]
+ domain = retgroups[0]["email"].split('@')[1]
domain_ending = domain.removeprefix(ctx.obj.data_partition + ".")
console.print(domain_ending)
else:
@@ -253,15 +265,17 @@ def groups(ctx: typer.Context,
if "policy" in str(grp):
console.print(grp)
+
def get_domain_suffix(ctx):
"""
get domain
"""
retgroups = request_groups(ctx)
- domain=retgroups[0]["email"].split('@')[1]
+ domain = retgroups[0]["email"].split('@')[1]
domain_ending = domain.removeprefix(ctx.obj.data_partition + ".")
return domain_ending
+
def is_json(myjson):
"""
Is this string valid json - detect problems before sending it to policy service
@@ -272,23 +286,26 @@ def is_json(myjson):
return False
return True
+
@cli.command(rich_help_panel="Policy Developer Utils")
def translate(ctx: typer.Context,
- policy_id: str= typer.Argument(...),
- output: OutputType = typer.Option("fancy", help="Output style"),
- file: Path = typer.Option(...,
- "-f", "--file",
- exists=True,
- file_okay=True,
- dir_okay=False,
- writable=False,
- readable=True,
- resolve_path=True,
- help="json file"
- ),
- force: bool = typer.Option(False, "--force", help="No confirmation prompt"),
- template: bool = typer.Option(False, "--template", "-t", help="Input is a template to be rendered")
-):
+ policy_id: str = typer.Argument(...),
+ output: OutputType = typer.Option("fancy", help="Output style"),
+ file: Path = typer.Option(...,
+ "-f", "--file",
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ writable=False,
+ readable=True,
+ resolve_path=True,
+ help="json file"
+ ),
+ force: bool = typer.Option(
+ False, "--force", help="No confirmation prompt"),
+ template: bool = typer.Option(
+ False, "--template", "-t", help="Input is a template to be rendered")
+ ):
"""
Translate testing utility.
@@ -310,26 +327,27 @@ def translate(ctx: typer.Context,
template_data = Template(data)
data = template_data.substitute(
{'data_partition': ctx.obj.data_partition,
- 'name': short_policy_id.removesuffix('.rego'),
- 'domain': domain
- })
+ 'name': short_policy_id.removesuffix('.rego'),
+ 'domain': domain
+ })
if output == output.fancy:
console.print(Panel(data, title="rendered data", highlight=True))
else:
print(data)
if not force:
- typer.confirm(f"translate above data against {short_policy_id} in {ctx.obj.data_partition}", abort=True)
+ typer.confirm(
+ f"translate above data against {short_policy_id} in {ctx.obj.data_partition}", abort=True)
if not is_json(data):
error_console.print("Error: Not valid json")
raise typer.Exit(1)
-
+
try:
with console.status("Translating..."):
r = requests.post(ctx.obj.url + "/translate",
- data=data,
- headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ data=data,
+ headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
error_console.print(f"Error: {e}")
raise SystemExit(e)
@@ -343,26 +361,29 @@ def translate(ctx: typer.Context,
error_console.print(f"Error: {r.json()['detail']} {r.status_code}")
else:
error_console.print(f"Error: {r.text}")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
@cli.command(rich_help_panel="Policy Commands")
def eval(ctx: typer.Context,
- policy_id: str= typer.Argument(None),
- file: Path = typer.Option(...,
- "-f", "--file",
- exists=True,
- file_okay=True,
- dir_okay=False,
- writable=False,
- readable=True,
- resolve_path=True,
- ),
- template: bool = typer.Option(False, "--template", "-t", help="Input is a template to be rendered"),
- output: OutputType = typer.Option("fancy", help="Output style"),
- force: bool = typer.Option(False, "--force", help="No confirmation prompt"),
- legal_tag: str = None,
- domain: str = None
- ):
+ policy_id: str = typer.Argument(None),
+ file: Path = typer.Option(...,
+ "-f", "--file",
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ writable=False,
+ readable=True,
+ resolve_path=True,
+ ),
+ template: bool = typer.Option(
+ False, "--template", "-t", help="Input is a template to be rendered"),
+ output: OutputType = typer.Option("fancy", help="Output style"),
+ force: bool = typer.Option(
+ False, "--force", help="No confirmation prompt"),
+ legal_tag: str = None,
+ domain: str = None
+ ):
"""
Evaluate Policy.
@@ -378,8 +399,8 @@ def eval(ctx: typer.Context,
This command will allow the policy service to add/replace any auth details in json using the include_auth capability in API.
"""
if policy_id.startswith('osdu/instance/'):
- error_console.print("Error: Instance policies not supported")
- raise typer.Abort()
+ error_console.print("Error: Instance policies not supported")
+ raise typer.Abort()
if domain is None:
domain = get_domain_suffix(ctx)
@@ -416,13 +437,15 @@ def eval(ctx: typer.Context,
raise typer.Exit(1)
if not force:
- typer.confirm(f"evaluate against {short_policy_id} in {ctx.obj.data_partition}", abort=True)
+ typer.confirm(
+ f"evaluate against {short_policy_id} in {ctx.obj.data_partition}", abort=True)
bdata = data.encode('utf-8')
files = {'file': (short_policy_id, bdata)}
- eval_result = evaluations_query(ctx = ctx, policy_id=policy_id, files=files)
+ eval_result = evaluations_query(ctx=ctx, policy_id=policy_id, files=files)
console.print(eval_result)
+
def search_policies_full(result: list, search: str):
"""
search helper
@@ -430,9 +453,10 @@ def search_policies_full(result: list, search: str):
retlist = []
for pol in result:
if re.search(search, str(pol)):
- retlist.append({'id':pol['id'], 'raw': pol['raw']})
+ retlist.append({'id': pol['id'], 'raw': pol['raw']})
return retlist
+
def search_policies_idonly(result: list, search: str):
"""
search helper in policy name only
@@ -440,29 +464,30 @@ def search_policies_idonly(result: list, search: str):
retlist = []
for pol in result:
if re.search(search, pol['id']):
- retlist.append({'id':pol['id'], 'raw': pol['raw']})
+ retlist.append({'id': pol['id'], 'raw': pol['raw']})
return retlist
-
+
+
def display_all_policies(ctx: typer.Context,
- output: OutputType = OutputType.fancy,
- raw: bool = False,
- search_name = None,
- search = None
- ):
+ output: OutputType = OutputType.fancy,
+ raw: bool = False,
+ search_name=None,
+ search=None
+ ):
"""
display policies
"""
-
+
# retrieve all policy ids
with console.status("Getting policies..."):
result = get_policies(ctx)
if search_name:
- with console.status("Searching policies by name..."):
+ with console.status("Searching policies by name..."):
result = search_policies_idonly(result, search_name)
if search:
- with console.status("Searching policies..."):
+ with console.status("Searching policies..."):
result = search_policies_full(result, search)
if result:
@@ -480,7 +505,7 @@ def display_all_policies(ctx: typer.Context,
instance_tree = tree.add("osdu/instance/")
data_partition_tree = partition_tree.add(ctx.obj.data_partition)
for pol in result:
- #pol['id'].basename
+ # pol['id'].basename
short_policy_id = os.path.basename(pol['id'])
dir = os.path.dirname(pol['id'])
if pol['id'].startswith("osdu/instance/"):
@@ -490,7 +515,7 @@ def display_all_policies(ctx: typer.Context,
console.print(tree)
raise typer.Exit()
- try:
+ try:
# silly math for figuring out terminal size
width, height = os.get_terminal_size()
except:
@@ -502,13 +527,13 @@ def display_all_policies(ctx: typer.Context,
for pol in result:
print(pol['id'])
raise typer.Exit()
-
+
# else let's make it look pretty
if search or search_name:
table = Table(title="Search results")
else:
table = Table(title="Lookup of all policy IDs")
- name_width=0
+ name_width = 0
for pol in result:
id_width = len(pol['id'])
@@ -517,28 +542,36 @@ def display_all_policies(ctx: typer.Context,
# silly math for figuring out terminal size
preview_width = width - 6 - name_width
- table.add_column("Name", justify="full", style="cyan", no_wrap=True, min_width=name_width)
- table.add_column("Preview", justify="left", style="green", no_wrap=True, max_width=preview_width)
+ table.add_column("Name", justify="full", style="cyan",
+ no_wrap=True, min_width=name_width)
+ table.add_column("Preview", justify="left", style="green",
+ no_wrap=True, max_width=preview_width)
for pol in result:
- preview=pol['raw'].replace("\n", " ").strip()
- preview=" ".join(preview.split())
+ preview = pol['raw'].replace("\n", " ").strip()
+ preview = " ".join(preview.split())
if preview.startswith("p"):
table.add_row(pol['id'], preview.strip())
else:
console.print("Error: unexpected preview data")
console.print(table)
+
@cli.command("ls", rich_help_panel="Policy Commands")
def list(ctx: typer.Context,
- policy_list: List[str]= typer.Argument(None),
- raw: bool = typer.Option(False, "--raw", help="Full output from policy service"),
- quiet: bool = typer.Option(False, "--quiet", help="Don't display policy"),
- output: OutputType = typer.Option("fancy", help="Output style"),
- search: str = typer.Option("", "--search", "-s", help="Regex search string", rich_help_panel="Search Options"),
- search_name: str = typer.Option("", "--name", "-n", help="Regex search string in policy name only", rich_help_panel="Search Options"),
- download: bool = typer.Option(False, "--download", "-d", help="download policy to file")
- ):
+ policy_list: List[str] = typer.Argument(None),
+ raw: bool = typer.Option(
+ False, "--raw", help="Full output from policy service"),
+ quiet: bool = typer.Option(
+ False, "--quiet", help="Don't display policy"),
+ output: OutputType = typer.Option("fancy", help="Output style"),
+ search: str = typer.Option(
+ "", "--search", "-s", help="Regex search string", rich_help_panel="Search Options"),
+ search_name: str = typer.Option(
+ "", "--name", "-n", help="Regex search string in policy name only", rich_help_panel="Search Options"),
+ download: bool = typer.Option(
+ False, "--download", "-d", help="download policy to file")
+ ):
'''
Lookup policies or a policy.
@@ -546,17 +579,19 @@ def list(ctx: typer.Context,
'''
if output == output.excel:
error_console.print("Error: Not supported output type")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
if policy_list == ['-']:
policy_list = sys.stdin.read().strip().split()
if not policy_list:
- display_all_policies(ctx = ctx, output = output, raw = raw, search=search, search_name=search_name)
+ display_all_policies(ctx=ctx, output=output, raw=raw,
+ search=search, search_name=search_name)
else:
for policy_id in policy_list:
if '*' in policy_id:
- display_all_policies(ctx = ctx, output = output, raw = raw, search=policy_id)
+ display_all_policies(ctx=ctx, output=output,
+ raw=raw, search=policy_id)
return
# if includes a name without .rego, let's add it to make it easier to use cli
if not policy_id.endswith(".rego"):
@@ -568,40 +603,50 @@ def list(ctx: typer.Context,
if result:
if raw:
if download:
- filename = short_policy_id.removesuffix('rego') + 'json'
+ filename = short_policy_id.removesuffix(
+ 'rego') + 'json'
with console.status(f"Saving {filename}..."):
f = open(filename, "w")
f.write(str(result))
f.close()
if not quiet:
- console.print(f"{policy_id} raw saved as '{filename}'")
+ console.print(
+ f"{policy_id} raw saved as '{filename}'")
elif quiet:
pass
elif output == output.fancy:
console.print(result)
- else: # simple
+ else: # simple
print(result)
- else:
+ else:
if download:
with console.status(f"Saving {short_policy_id}..."):
f = open(short_policy_id, "w")
f.write(str(result['raw']))
f.close()
if not quiet:
- console.print(f"{policy_id} saved as '{short_policy_id}'")
+ console.print(
+ f"{policy_id} saved as '{short_policy_id}'")
elif quiet:
pass
elif output == output.fancy:
console.print(result['raw'])
- else: # simple
+ else: # simple
print(result['raw'])
else:
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
+
+def cleanup_url(url):
+ if url.endswith("/"):
+ return url.rstrip(url[-1])
+ return url
+
@cli.command(hidden=True, rich_help_panel="Policy Developer Utils")
def setup(ctx: typer.Context,
- raw: bool = typer.Option(False, "--raw", help="Don't Mask details"),
- ):
+ raw: bool = typer.Option(False, "--raw", help="Don't Mask details"),
+ ):
"""
Display setup details. Hidden command
"""
@@ -612,20 +657,26 @@ def setup(ctx: typer.Context,
else:
x = dict(sorted(vars(ctx.obj).items()))
for key in x:
- if "token" in key:
- console.print(f"[green]{key}[/]: ****")
- else:
- if key.endswith('/'):
- console.print(f"[green]{key}[/]: {x[key]} [red]Warning ending /[/]")
+ if "token" in key:
+ console.print(f"[green]{key}[/]: ****")
+ else:
+ if str(x[key]).endswith("/"):
+ console.print(
+ f"[green]{key}[/]: {x[key]} [red]Warning ending /[/]")
else:
console.print(f"[green]{key}[/]: {x[key]}")
-
+
for name, value in os.environ.items():
- if "token" in name.lower() or "key" in name.lower():
+ if "token" in name.lower() or \
+ "key" in name.lower() or \
+ "password" in name.lower() or \
+ "secret" in name.lower() or \
+ "client" in name.lower():
console.print("{0}: {1}".format(name, "****"))
else:
console.print("{0}: {1}".format(name, value))
+
@cli.command(rich_help_panel="Policy Commands")
def health(ctx: typer.Context):
"""
@@ -634,19 +685,21 @@ def health(ctx: typer.Context):
try:
with console.status("Getting health..."):
r = requests.get(ctx.obj.url + "/health", headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
- console.print(":thumbsup-emoji:",r.text)
+ console.print(":thumbsup-emoji:", r.text)
else:
error_console.print(r.text)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
@cli.command(rich_help_panel="Utils")
def info(ctx: typer.Context,
- service: str = typer.Option("policy", "-s", "--service", help="Get search info")
- ):
+ service: str = typer.Option(
+ "policy", "-s", "--service", help="Get search info")
+ ):
"""
Info on Dataset, Entitlement, Legal, Policy Service, Search, and Storage
"""
@@ -664,14 +717,16 @@ def info(ctx: typer.Context,
get_info(ctx=ctx, base_url=ctx.obj.dataset_url)
else:
error_console.print(f"Error: Unsupported service {service}")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
@cli.command(rich_help_panel="Policy Developer Utils")
def diff(ctx: typer.Context,
- policy_id_1: str = typer.Argument(...),
- policy_id_2: str = typer.Argument(...),
- n: int = typer.Option(3, "-n", help="Number of adjacent lines to show"),
- ):
+ policy_id_1: str = typer.Argument(...),
+ policy_id_2: str = typer.Argument(...),
+ n: int = typer.Option(
+ 3, "-n", help="Number of adjacent lines to show"),
+ ):
"""
Compare two policies, show the delta in a context diff format.
@@ -698,46 +753,51 @@ def diff(ctx: typer.Context,
n=n)
console.print('\n'.join(differences))
+
def lookup_policy(ctx: typer.Context, policy_id, quiet=False):
short_policy_id = os.path.basename(policy_id)
- if "osdu/instance/" in policy_id: # lookup instance policy
+ if "osdu/instance/" in policy_id: # lookup instance policy
with console.status(f"Getting instance policy {short_policy_id}..."):
result = get_instance_policy(ctx, short_policy_id, quiet)
- elif "osdu/partition/" in policy_id: # lookup partition policy
+ elif "osdu/partition/" in policy_id: # lookup partition policy
with console.status(f"Getting partition policy {short_policy_id}..."):
result = get_partition_policy(ctx, short_policy_id, quiet)
- else: # assume it's a partition policy
+ else: # assume it's a partition policy
with console.status(f"Getting partition policy {short_policy_id}..."):
result = get_partition_policy(ctx, short_policy_id, quiet)
if not result:
# nope report the error and let's still check for a instance policy
if not quiet:
- error_console.print(f"[green]Checking for '{short_policy_id}' in instance policies[/]...")
+ error_console.print(
+ f"[green]Checking for '{short_policy_id}' in instance policies[/]...")
result = get_instance_policy(ctx, short_policy_id, quiet)
if not result:
if not quiet:
error_console.print(f"Error: Unable to find {policy_id}")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
return result
+
@cli.command(rich_help_panel="Policy Developer Utils")
def compile(ctx: typer.Context,
- file: Path = typer.Option(...,
- "-f", "--file",
- exists=True,
- file_okay=True,
- dir_okay=False,
- writable=False,
- readable=True,
- resolve_path=True,
- ),
- metrics: bool = False,
- opt_walk: bool = typer.Option(False, "--walk"),
- instrument: bool = False,
- template: bool = typer.Option(False, "--template", "-t", help="Input is a template to be rendered"),
- force: bool = typer.Option(False, "--force", help="No confirmation prompt")
- ):
+ file: Path = typer.Option(...,
+ "-f", "--file",
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ writable=False,
+ readable=True,
+ resolve_path=True,
+ ),
+ metrics: bool = False,
+ opt_walk: bool = typer.Option(False, "--walk"),
+ instrument: bool = False,
+ template: bool = typer.Option(
+ False, "--template", "-t", help="Input is a template to be rendered"),
+ force: bool = typer.Option(
+ False, "--force", help="No confirmation prompt")
+ ):
"""
Use OPA's Compile API to partially evaluate a query.
@@ -760,45 +820,50 @@ def compile(ctx: typer.Context,
params = {'metrics': metrics, 'instrument': instrument}
try:
- r = requests.post(ctx.obj.url + "/compile", files=files, params=params, headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ r = requests.post(ctx.obj.url + "/compile", files=files,
+ params=params, headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
console.print(r.json())
else:
error_console.print(r.text, r.status_code)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
if opt_walk:
# Load the resulting set of query ASTs out of the JSON response.
if "result" in r.json() and "queries" in r.json()["result"]:
tree = r.json()["result"]["queries"]
- console.print(Panel(str(tree), title="Abstract Syntax Tree", highlight=True))
+ console.print(
+ Panel(str(tree), title="Abstract Syntax Tree", highlight=True))
qs = ast.QuerySet.from_data(tree)
# Pretty print the ASTs.
walk.pretty_print(qs)
else:
- error_console.print("Error: unexpected response from policy service. Check if policy exists")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ "Error: unexpected response from policy service. Check if policy exists")
+ raise typer.Exit(1) # non-zero exit status
+
@cli.command(rich_help_panel="Policy Developer Utils")
def config(ctx: typer.Context):
"""
Diagnostic config on Policy Service.
-
+
Requires service have ENABLE_DEV_DIAGNOSTICS enabled.
"""
try:
r = requests.get(ctx.obj.diag_url + "/config", headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
console.print(r.json())
else:
error_console.print(r.text)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def get_legal_tags(ctx: typer.Context, tag_name: str = None):
"""
@@ -807,18 +872,21 @@ def get_legal_tags(ctx: typer.Context, tag_name: str = None):
try:
if tag_name:
with console.status(f"Getting legal tag {tag_name}..."):
- r = requests.get(ctx.obj.legal_url + "/legaltags/" + tag_name, headers = headers(ctx))
+ r = requests.get(ctx.obj.legal_url +
+ "/legaltags/" + tag_name, headers=headers(ctx))
else:
with console.status("Getting legaltags..."):
- r = requests.get(ctx.obj.legal_url + "/legaltags" + "?valid=true", headers = headers(ctx))
- except requests.exceptions.RequestException as e:
+ r = requests.get(ctx.obj.legal_url + "/legaltags" +
+ "?valid=true", headers=headers(ctx))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
-
+
if r.ok:
return r.json()
else:
error_console.print("Error:", r.text)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def create_legal_tag(ctx: typer.Context, json_data, update=False):
"""
@@ -827,22 +895,25 @@ def create_legal_tag(ctx: typer.Context, json_data, update=False):
try:
if update:
with console.status(f"Updating legal tag..."):
- r = requests.put(ctx.obj.legal_url + "/legaltags", json=json.loads(json_data), headers = headers(ctx, content_type_json=True))
+ r = requests.put(ctx.obj.legal_url + "/legaltags", json=json.loads(
+ json_data), headers=headers(ctx, content_type_json=True))
else:
with console.status(f"Creating legal tag..."):
- r = requests.post(ctx.obj.legal_url + "/legaltags", json=json.loads(json_data), headers = headers(ctx, content_type_json=True))
- except requests.exceptions.RequestException as e:
+ r = requests.post(ctx.obj.legal_url + "/legaltags", json=json.loads(
+ json_data), headers=headers(ctx, content_type_json=True))
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
-
+
if r.ok:
return r.json()
else:
if r.status_code == 409:
error_console.print("Error: Legal tag already exists")
- raise typer.Exit(2) # non-zero exit status
+ raise typer.Exit(2) # non-zero exit status
else:
error_console.print("Error:", r.text, r.status_code)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def get_a_random_legal_tag(ctx: typer.Context):
"""
@@ -852,19 +923,21 @@ def get_a_random_legal_tag(ctx: typer.Context):
lucky_tag = random.choice(r_json["legalTags"])
return lucky_tag["name"].strip()
+
@cli.command(rich_help_panel="Utils")
def add_legal_tag(ctx: typer.Context,
- file: Path = typer.Option(...,
- "-f", "--file",
- exists=True,
- file_okay=True,
- dir_okay=False,
- writable=False,
- readable=True,
- resolve_path=True,
- ),
- update: bool = typer.Option(False, "--update", help="Update existing legal tag")
- ):
+ file: Path = typer.Option(...,
+ "-f", "--file",
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ writable=False,
+ readable=True,
+ resolve_path=True,
+ ),
+ update: bool = typer.Option(
+ False, "--update", help="Update existing legal tag")
+ ):
"""
Create or Update a legal tag
@@ -882,16 +955,20 @@ def add_legal_tag(ctx: typer.Context,
r_json = create_legal_tag(ctx=ctx, json_data=data, update=update)
console.print(json.dumps(r_json, indent=4))
+
@cli.command(rich_help_panel="Utils")
def legal_tags(ctx: typer.Context,
- tag_list: List[str]= typer.Argument(None, help="List of tags to show"),
- raw: bool = typer.Option(False, "--raw", help="Get raw json output"),
- random: bool = typer.Option(False, "--random", help="Get a random legal tag"),
- limit: int = typer.Option(None, help="Limit number")):
+ tag_list: List[str] = typer.Argument(
+ None, help="List of tags to show"),
+ raw: bool = typer.Option(
+ False, "--raw", help="Get raw json output"),
+ random: bool = typer.Option(
+ False, "--random", help="Get a random legal tag"),
+ limit: int = typer.Option(None, help="Limit number")):
"""
Show legal tags.
"""
-
+
if tag_list == ['-']:
tag_list = sys.stdin.read().strip().split()
@@ -914,35 +991,40 @@ def legal_tags(ctx: typer.Context,
if count == limit:
break
+
@cli.command("opa-add", rich_help_panel="Policy Developer Utils")
def add_to_opa(ctx: typer.Context,
- policy_id: str = typer.Argument(...),
- file: Path = typer.Option(...,
- "-f", "--file",
- exists=True,
- file_okay=True,
- dir_okay=False,
- writable=False,
- readable=True,
- resolve_path=True,
- ),
- template: bool = typer.Option(False, "--template", "-t", help="Input is a template to be rendered"),
- force: bool = typer.Option(False, "--force", help="No confirmation prompt"),
- url: str = typer.Option("http://localhost:8181", help="Base URL to connect to OPA")
- ):
+ policy_id: str = typer.Argument(...),
+ file: Path = typer.Option(...,
+ "-f", "--file",
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ writable=False,
+ readable=True,
+ resolve_path=True,
+ ),
+ template: bool = typer.Option(
+ False, "--template", "-t", help="Input is a template to be rendered"),
+ force: bool = typer.Option(
+ False, "--force", help="No confirmation prompt"),
+ url: str = typer.Option(
+ "http://localhost:8181", help="Base URL to connect to OPA")
+ ):
"""
[green]Add or update[/green] a policy directly in OPA :sparkles: for LOCAL testing/development
"""
if ctx.obj.debug:
console.print(f"url: {url}")
-
+
if policy_id.startswith('osdu/instance/'):
policy_type = "instance"
elif policy_id.startswith('osdu/partition/'):
policy_type = "partition"
else:
- error_console.print("Error: policy_id must start with osdu/instance/ or osdu/partition/")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ "Error: policy_id must start with osdu/instance/ or osdu/partition/")
+ raise typer.Exit(1) # non-zero exit status
if not policy_id.endswith(".rego"):
policy_id = policy_id + ".rego"
@@ -959,37 +1041,43 @@ def add_to_opa(ctx: typer.Context,
template_data = Template(data)
data = template_data.substitute(
{
- 'data_partition': ctx.obj.data_partition,
- 'DATA_PARTITION': ctx.obj.data_partition,
- 'name': short_policy_id.removesuffix('.rego')
+ 'data_partition': ctx.obj.data_partition,
+ 'DATA_PARTITION': ctx.obj.data_partition,
+ 'name': short_policy_id.removesuffix('.rego')
}
)
- console.print(Panel(data, title=f"rendered {policy_type} policy", highlight=True))
+ console.print(
+ Panel(data, title=f"rendered {policy_type} policy", highlight=True))
if not force:
typer.confirm(f"Add {policy_id} to OPA", abort=True)
-
+
with console.status(f"Adding policy {policy_id}"):
- r = opa.put_opa_policy_direct(policy_id=policy_id, data=data, base_url=url)
+ r = opa.put_opa_policy_direct(
+ policy_id=policy_id, data=data, base_url=url)
if r.ok:
console.print(f"Policy {policy_id} added to OPA")
else:
- error_console.print(f"Error adding policy {policy_id} to OPA {r.json()} {r.status_code}")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ f"Error adding policy {policy_id} to OPA {r.json()} {r.status_code}")
+ raise typer.Exit(1) # non-zero exit status
+
@cli.command("check", rich_help_panel="Policy Developer Utils")
def opa_check(
- ctx: typer.Context,
- file_list: List[str]= typer.Argument(...),
- display: bool = typer.Option(False, "--display", "-d", help="Display policy"),
- template: bool = typer.Option(False, "--template", "-t", help="Input is a template to be rendered"),
- ):
+ ctx: typer.Context,
+ file_list: List[str] = typer.Argument(...),
+ display: bool = typer.Option(
+ False, "--display", "-d", help="Display policy"),
+ template: bool = typer.Option(
+ False, "--template", "-t", help="Input is a template to be rendered"),
+):
"""
Check rego file for errors
"""
if file_list == ['-']:
file_list = sys.stdin.read().strip().split()
-
+
chk = opa.OpaCheck()
for file in file_list:
policy_id = os.path.basename(file)
@@ -1018,22 +1106,26 @@ def opa_check(
result, error_msg = chk.check(rego=data, filename=policy_id)
if result:
if 'rego_parse_error' in error_msg:
- error_console.print(f"{policy_id}: rego_parse_error:\n{error_msg}")
+ error_console.print(
+ f"{policy_id}: rego_parse_error:\n{error_msg}")
else:
error_console.print(f"{policy_id}: error {error_msg}")
- chk.delete() # cleanup before exit
- raise typer.Exit(1) # non-zero exit status
+ chk.delete() # cleanup before exit
+ raise typer.Exit(1) # non-zero exit status
else:
console.print(f"{policy_id}: [green]OK[/]")
chk.delete()
+
@cli.command("opa-rm", rich_help_panel="Policy Developer Utils")
def delete_from_opa(
- ctx: typer.Context,
- policy_list: List[str]= typer.Argument(...),
- force: bool = typer.Option(False, "--force", help="No confirmation prompt"),
- url: str = typer.Option("http://localhost:8181", help="Base URL to connect to OPA")
- ):
+ ctx: typer.Context,
+ policy_list: List[str] = typer.Argument(...),
+ force: bool = typer.Option(
+ False, "--force", help="No confirmation prompt"),
+ url: str = typer.Option("http://localhost:8181",
+ help="Base URL to connect to OPA")
+):
"""
[red]delete[/red] a policy directly from OPA. :fire: for LOCAL testing/development
@@ -1051,8 +1143,9 @@ def delete_from_opa(
elif policy_id.startswith('osdu/partition/'):
policy_type = "partition"
else:
- error_console.print("Error: policy_id must start with osdu/instance/ or osdu/partition/")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ "Error: policy_id must start with osdu/instance/ or osdu/partition/")
+ raise typer.Exit(1) # non-zero exit status
if not policy_id.endswith(".rego"):
policy_id = policy_id + ".rego"
@@ -1067,26 +1160,40 @@ def delete_from_opa(
if r.ok:
console.print(f"Policy {policy_id} deleted from OPA")
else:
- error_console.print(f"Error deleting policy {policy_id} from OPA {r.json()} {r.status_code}")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ f"Error deleting policy {policy_id} from OPA {r.json()} {r.status_code}")
+ raise typer.Exit(1) # non-zero exit status
+
@cli.callback()
def main(
ctx: typer.Context,
token: str = typer.Option(None, '-t', '--token', envvar="TOKEN"),
- policy_url: str = typer.Option(None, '-p', '--policy-url', envvar="POLICY_URL"),
- base_url: str = typer.Option(None, '--host', '--url', '--base-url', envvar="BASE_URL"),
- entitlements_url: str = typer.Option(None, '-e', '--ent-url', envvar="ENTITLEMENTS_URL", hidden=True),
- storage_url: str = typer.Option(None, '-s', '--storage-url', envvar="STORAGE_URL", hidden=True),
- search_url: str = typer.Option(None, '-x', '--ent-url', envvar="SEARCH_URL", hidden=True),
- legal_url: str = typer.Option(None, '-l', '--legal-url', envvar="LEGAL_URL", hidden=True),
- dataset_url: str = typer.Option(None, '--dataset-url', envvar="DATASET_URL", hidden=True),
- path: str = typer.Option('/api/policy/v1', envvar="POLICY_PATH", hidden=True),
- data_partition: str = typer.Option(None, '-d', '--data-partition-id', envvar="DATA_PARTITION"),
- x_user_id: str = typer.Option(None, envvar="XUSERID", help="optional user id, added to headers"),
+ policy_url: str = typer.Option(
+ None, '-p', '--policy-url', envvar="POLICY_URL"),
+ base_url: str = typer.Option(
+ None, '--host', '--url', '--base-url', envvar="BASE_URL"),
+ entitlements_url: str = typer.Option(
+ None, '-e', '--ent-url', envvar="ENTITLEMENTS_BASE_URL", hidden=True),
+ storage_url: str = typer.Option(
+ None, '-s', '--storage-url', envvar="STORAGE_BASE_URL", hidden=True),
+ search_url: str = typer.Option(
+ None, '-x', '--ent-url', envvar="SEARCH_BASE_URL", hidden=True),
+ legal_url: str = typer.Option(
+ None, '-l', '--legal-url', envvar="LEGAL_BASE_URL", hidden=True),
+ dataset_url: str = typer.Option(
+ None, '--dataset-url', envvar="DATASET_BASE_URL", hidden=True),
+ path: str = typer.Option(
+ '/api/policy/v1', envvar="POLICY_PATH", hidden=True),
+ data_partition: str = typer.Option(
+ None, '-d', '--data-partition-id', envvar="DATA_PARTITION"),
+ x_user_id: str = typer.Option(
+ None, envvar="XUSERID", help="optional user id, added to headers"),
debug: bool = typer.Option(False, hidden=True, help="Debug output"),
- correlation_id: bool = typer.Option(True, hidden=True, help="Enable/disable correlation-id"),
- verbose: bool = typer.Option(False, "-V", "--verbose", hidden=True, help="Verbose"),
+ correlation_id: bool = typer.Option(
+ True, hidden=True, help="Enable/disable correlation-id"),
+ verbose: bool = typer.Option(
+ False, "-V", "--verbose", hidden=True, help="Verbose"),
version: Optional[bool] = typer.Option(
None,
"--version",
@@ -1095,22 +1202,25 @@ def main(
callback=_version_callback,
is_eager=True
)
- ) -> None:
+) -> None:
if not token:
error_console.print("Missing token; pass --token or set env[TOKEN]")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
if not data_partition:
- error_console.print("Missing data_partition; pass --data-partition or set env[DATA_PARTITION]")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ "Missing data_partition; pass --data-partition or set env[DATA_PARTITION]")
+ raise typer.Exit(1) # non-zero exit status
if not base_url:
if not policy_url:
- error_console.print("Missing url and base_url; pass --url or set env[POLICY_URL]")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ "Missing url and base_url; pass --url or set env[POLICY_URL]")
+ raise typer.Exit(1) # non-zero exit status
base_url = policy_url
+ base_url = cleanup_url(base_url)
if not policy_url:
policy_url = base_url
@@ -1130,26 +1240,26 @@ def main(
dataset_url = base_url + "/api/dataset/v1"
uuid = None
- if correlation_id:
+ if correlation_id:
uuid = uuid7str()
ctx.obj = SimpleNamespace(
- token = token,
- data_partition = data_partition,
- url = policy_url + path,
- base_url = base_url,
- diag_url = policy_url + "/diag",
- entitlements_url = entitlements_url,
- storage_url = storage_url,
- search_url = search_url,
- dataset_url = dataset_url,
- debug = debug,
- legal_url = legal_url,
- uuid = uuid,
- user_id = getpass.getuser(),
- x_user_id = x_user_id)
-
- if debug: # hidden option
+ token=token,
+ data_partition=data_partition,
+ url=cleanup_url(policy_url) + path,
+ base_url=base_url,
+ diag_url=cleanup_url(policy_url) + "/diag",
+ entitlements_url=cleanup_url(entitlements_url),
+ storage_url=cleanup_url(storage_url),
+ search_url=cleanup_url(search_url),
+ dataset_url=cleanup_url(dataset_url),
+ debug=debug,
+ legal_url=cleanup_url(legal_url),
+ uuid=uuid,
+ user_id=getpass.getuser(),
+ x_user_id=x_user_id)
+
+ if debug: # hidden option
console.print(f"token: {ctx.obj.token}")
console.print(f"base_url: {ctx.obj.base_url}")
console.print(f"policy_url: {ctx.obj.url}")
@@ -1158,28 +1268,31 @@ def main(
console.print(f"search_url: {ctx.obj.search_url}")
console.print(f"legal_url: {ctx.obj.legal_url}")
console.print(f"data_partition: {ctx.obj.data_partition}")
- console.print(f"dataset_url: {ctx.obj.dataset_url}")
+ console.print(f"dataset_url: {ctx.obj.dataset_url}")
console.print(f"x-user-id: {ctx.obj.x_user_id}")
console.print(f"uuid: {ctx.obj.uuid}")
if verbose:
console.print(f"Correlation-ID: {ctx.obj.uuid}")
+
@cli.command(rich_help_panel="Policy Commands")
def add(ctx: typer.Context,
- policy_id: str = typer.Argument(...),
- #file: typer.FileBinaryRead = typer.Option(..., encoding='utf-8')
- file: Path = typer.Option(...,
- "-f", "--file",
- exists=True,
- file_okay=True,
- dir_okay=False,
- writable=False,
- readable=True,
- resolve_path=True,
- ),
- template: bool = typer.Option(False, "--template", "-t", help="Input is a template to be rendered"),
- force: bool = typer.Option(False, "--force", help="No confirmation prompt")
- ):
+ policy_id: str = typer.Argument(...),
+ #file: typer.FileBinaryRead = typer.Option(..., encoding='utf-8')
+ file: Path = typer.Option(...,
+ "-f", "--file",
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ writable=False,
+ readable=True,
+ resolve_path=True,
+ ),
+ template: bool = typer.Option(
+ False, "--template", "-t", help="Input is a template to be rendered"),
+ force: bool = typer.Option(
+ False, "--force", help="No confirmation prompt")
+ ):
"""
[green]Add or update[/green] a policy. :sparkles:
@@ -1194,7 +1307,8 @@ def add(ctx: typer.Context,
short_policy_id = os.path.basename(policy_id)
if not short_policy_id.endswith(".rego"):
short_policy_id = short_policy_id + ".rego"
- console.print(f"add contents of {file} as '{short_policy_id}' in {ctx.obj.data_partition}")
+ console.print(
+ f"add contents of {file} as '{short_policy_id}' in {ctx.obj.data_partition}")
with rich.progress.open(file, "r") as f:
data = f.read()
f.close()
@@ -1203,26 +1317,28 @@ def add(ctx: typer.Context,
template_data = Template(data)
data = template_data.substitute(
{'data_partition': ctx.obj.data_partition,
- 'DATA_PARTITION': ctx.obj.data_partition,
- 'name': short_policy_id.removesuffix('.rego')
- })
+ 'DATA_PARTITION': ctx.obj.data_partition,
+ 'name': short_policy_id.removesuffix('.rego')
+ })
console.print(Panel(data, title="rendered policy", highlight=True))
if not force:
- typer.confirm(f"add {short_policy_id} in {ctx.obj.data_partition}", abort=True)
+ typer.confirm(
+ f"add {short_policy_id} in {ctx.obj.data_partition}", abort=True)
bdata = data.encode('utf-8')
files = {'file': (short_policy_id, bdata)}
with console.status(f"Adding partition policy {short_policy_id}"):
add_partition_policy(ctx=ctx, policy_id=short_policy_id, files=files)
+
@cli.command("rm", rich_help_panel="Policy Commands")
def delete(
- ctx: typer.Context,
- policy_list: List[str]= typer.Argument(...),
- force: bool = typer.Option(False, "--force", help="No confirmation prompt")
- #... if sys.stdin.isatty() else sys.stdin.read().strip()
- ):
+ ctx: typer.Context,
+ policy_list: List[str] = typer.Argument(...),
+ force: bool = typer.Option(False, "--force", help="No confirmation prompt")
+ #... if sys.stdin.isatty() else sys.stdin.read().strip()
+):
"""
[red]delete[/red] a partition policy. :fire:
@@ -1249,9 +1365,11 @@ def delete(
if result:
long_policy_id = result['id']
if not force and interactive:
- delete = typer.confirm(f"Delete {long_policy_id} in {ctx.obj.data_partition}", abort=True)
+ delete = typer.confirm(
+ f"Delete {long_policy_id} in {ctx.obj.data_partition}", abort=True)
elif not force and not interactive:
- console.print(f"preview: would have deleted '{long_policy_id} in {ctx.obj.data_partition}'. Use --force")
+ console.print(
+ f"preview: would have deleted '{long_policy_id} in {ctx.obj.data_partition}'. Use --force")
continue
if delete or force:
@@ -1267,12 +1385,14 @@ def delete(
with console.status(f"Deleting partition policy {short_policy_id}"):
result = delete_partition_policy(ctx, short_policy_id)
if result:
- console.print(f"Policy [green]{short_policy_id}[/] deleted")
+ console.print(
+ f"Policy [green]{short_policy_id}[/] deleted")
else:
error_console.print(f"Delete of {short_policy_id} failed")
else:
error_console.print(f"Skipping delete of {policy_id}")
+
def get_info(ctx: typer.Context, base_url):
"""
Get info
@@ -1280,25 +1400,31 @@ def get_info(ctx: typer.Context, base_url):
try:
with console.status("Getting info..."):
r = requests.get(base_url + "/info", headers=headers(ctx))
- except requests.exceptions.RequestException as e:
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
console.print(r.json())
else:
error_console.print(r.text)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
@cli.command(rich_help_panel="Utils")
def storage(ctx: typer.Context,
- id_list: List[str]= typer.Argument(..., help="IDs to retrieve"),
- versions: bool = typer.Option(False, "--versions", "-V", help="Show versions"),
- version: str = typer.Option(None, "-v", "--version", help="Get particular version"),
- dataset: bool = typer.Option(False, "--dataset", "-D", help="dataset"),
- get: bool = typer.Option(False, "--get", hidden=True, help="get dataset"),
- download: bool = typer.Option(False, "--download", "-d", help="download dataset"),
- raw: bool = typer.Option(False, "--raw", help="json output"),
- ):
+ id_list: List[str] = typer.Argument(..., help="IDs to retrieve"),
+ versions: bool = typer.Option(
+ False, "--versions", "-V", help="Show versions"),
+ version: str = typer.Option(
+ None, "-v", "--version", help="Get particular version"),
+ dataset: bool = typer.Option(
+ False, "--dataset", "-D", help="dataset"),
+ get: bool = typer.Option(
+ False, "--get", hidden=True, help="get dataset"),
+ download: bool = typer.Option(
+ False, "--download", "-d", help="download dataset"),
+ raw: bool = typer.Option(False, "--raw", help="json output"),
+ ):
"""
Storage and Dataset record retrieval utility
"""
@@ -1307,7 +1433,7 @@ def storage(ctx: typer.Context,
if not len(id_list):
error_console.print("Error: Missing input 'ID_LIST...'")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
for id in id_list:
try:
@@ -1325,11 +1451,11 @@ def storage(ctx: typer.Context,
with console.status(f"Retrieving record..."):
r = requests.get(url,
- headers=headers(ctx)
- )
- except requests.exceptions.RequestException as e:
+ headers=headers(ctx)
+ )
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
-
+
if r.ok:
if raw:
console.print(r.json())
@@ -1339,30 +1465,33 @@ def storage(ctx: typer.Context,
console.print(r.json())
else:
error_console.print("Error:", r.text, r.status_code)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def get_filename_from_storage(ctx: typer.Context, id):
try:
with console.status(f"Retrieving storage record {id}..."):
r = requests.get(ctx.obj.storage_url + "/records/" + id,
- headers=headers(ctx)
- )
- except requests.exceptions.RequestException as e:
+ headers=headers(ctx)
+ )
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
-
+
if r.ok:
r_json = r.json()
if "data" in r_json and \
"DatasetProperties" in r_json["data"] and \
"FileSourceInfo" in r_json["data"]["DatasetProperties"] and \
- "Name" in r_json["data"]["DatasetProperties"]["FileSourceInfo"]:
+ "Name" in r_json["data"]["DatasetProperties"]["FileSourceInfo"]:
name = r_json["data"]["DatasetProperties"]["FileSourceInfo"]["Name"]
return name
else:
- error_console.print(f"Error: FileSourceInfo Name Path missing ['data']['DatasetProperties']['FileSourceInfo']['Name'] in record {id}")
+ error_console.print(
+ f"Error: FileSourceInfo Name Path missing ['data']['DatasetProperties']['FileSourceInfo']['Name'] in record {id}")
if ctx.obj.debug:
console.print(r_json)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
def get_dataset(ctx: typer.Context, data, download=False, raw=False):
id = data["delivery"][0]["datasetRegistryId"]
@@ -1370,12 +1499,12 @@ def get_dataset(ctx: typer.Context, data, download=False, raw=False):
name = get_filename_from_storage(ctx=ctx, id=id)
filename = os.path.basename(name)
signed_url = data["delivery"][0]["retrievalProperties"]["signedUrl"]
-
+
if download:
try:
with console.status(f"Downloading {id} {name}..."):
r = requests.get(signed_url)
- except requests.exceptions.RequestException as e:
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
if r.ok:
@@ -1388,49 +1517,63 @@ def get_dataset(ctx: typer.Context, data, download=False, raw=False):
console.print(f"unsigned_url: {unsigned_url}")
console.print(f"signed_url: {signed_url}")
+
@cli.command(rich_help_panel="Utils")
def search(ctx: typer.Context,
- id: str = typer.Argument(None),
- kind: str = typer.Option("*:*:*:*", help="The kind(s) of the record to query"),
- query: str = typer.Option("", help="Query string based on Lucene query string syntax"),
- sort_field: str = typer.Option("id", help="Sort field"),
- sort_order: str = typer.Option("ASC", help="Sort Order", callback=search_cli.search_order_callback),
- limit: int = typer.Option(10, help="The maximum number of results to return from the given offset"),
- offset: int = typer.Option(0, help="The starting offset from which to return results"),
- detail: bool = typer.Option(False, "--detail", help="Show more details"),
- id_only: bool = typer.Option(False, "--id-only", help="Only show IDs"),
- as_owner: bool = typer.Option(False, "--as-owner", help="Query as owner"),
- random_id: bool = typer.Option(False, "--random-id", help="Get a random ID"),
- output: OutputType = typer.Option("fancy", help="Output style"),
- file: Path = typer.Option("search.xlsx",
- "-f", "--file",
- dir_okay=False,
- resolve_path=True,
- help="Excel file for output=excel"
- ),
- open_file: bool = typer.Option(False, "--open", help="Open file after saving"),
- raw: bool = typer.Option(False, "--raw", help="Show raw/json search results"),
- spatial_filter: Path = typer.Option(None,
- exists=True,
- file_okay=True,
- dir_okay=False,
- writable=False,
- readable=True,
- resolve_path=True,
- help="Spatial filter json file"
- ),
- with_cursor: bool = typer.Option(False, "--with-cursor", help="Use cursors"),
- cursor: str = typer.Option(None, "--cursor", help="Pass cursor back")
- ):
+ id: str = typer.Argument(None),
+ kind: str = typer.Option(
+ "*:*:*:*", help="The kind(s) of the record to query"),
+ query: str = typer.Option(
+ "", help="Query string based on Lucene query string syntax"),
+ sort_field: str = typer.Option("id", help="Sort field"),
+ sort_order: str = typer.Option(
+ "ASC", help="Sort Order", callback=search_cli.search_order_callback),
+ limit: int = typer.Option(
+ 10, help="The maximum number of results to return from the given offset"),
+ offset: int = typer.Option(
+ 0, help="The starting offset from which to return results"),
+ detail: bool = typer.Option(
+ False, "--detail", help="Show more details"),
+ id_only: bool = typer.Option(
+ False, "--id-only", help="Only show IDs"),
+ as_owner: bool = typer.Option(
+ False, "--as-owner", help="Query as owner"),
+ random_id: bool = typer.Option(
+ False, "--random-id", help="Get a random ID"),
+ output: OutputType = typer.Option("fancy", help="Output style"),
+ file: Path = typer.Option("search.xlsx",
+ "-f", "--file",
+ dir_okay=False,
+ resolve_path=True,
+ help="Excel file for output=excel"
+ ),
+ open_file: bool = typer.Option(
+ False, "--open", help="Open file after saving"),
+ raw: bool = typer.Option(
+ False, "--raw", help="Show raw/json search results"),
+ spatial_filter: Path = typer.Option(None,
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ writable=False,
+ readable=True,
+ resolve_path=True,
+ help="Spatial filter json file"
+ ),
+ with_cursor: bool = typer.Option(
+ False, "--with-cursor", help="Use cursors"),
+ cursor: str = typer.Option(
+ None, "--cursor", help="Pass cursor back")
+ ):
"""
Search utility
Kind must follow the convention: {Schema-Authority}:{dataset-name}:{record-type}:{version}
-
+
Example Searches:
Wellbore Master Data Instances for Well with ID 1691:
--kind="*:*:master-data--Wellbore:*" --query=data.WellID:\\"osdu:master-data--Well:1691:\\"
-
+
Wellbore Trajectory Work Product Components associated with Wellbore ID 1691:
--kind="*:*:work-product-component--WellboreTrajectory:*" --query=data.WellboreID:\\"osdu:master-data--Wellbore:1691:\\"
@@ -1458,7 +1601,7 @@ def search(ctx: typer.Context,
The cursor will expire after 1 min and won't return any more results if the requests are not made in specified time.
"""
if id:
- query=f"id:\"{id}\""
+ query = f"id:\"{id}\""
# Note: above \\ is for help output, so just one \ is needed
@@ -1481,8 +1624,8 @@ def search(ctx: typer.Context,
# }
sort_data = {
- 'field': [ sort_field ],
- 'order': [ sort_order ]
+ 'field': [sort_field],
+ 'order': [sort_order]
}
search_json_data = {
'kind': kind,
@@ -1499,8 +1642,9 @@ def search(ctx: typer.Context,
spatial_filter_data = json.load(f)
f.close()
except json.decoder.JSONDecodeError as err:
- error_console.print(f"Error: {spatial_filter} malformed json. {err}")
- raise typer.Exit(1) # non-zero exit status
+ error_console.print(
+ f"Error: {spatial_filter} malformed json. {err}")
+ raise typer.Exit(1) # non-zero exit status
search_json_data["spatialFilter"] = spatial_filter_data
#search_json_data["returnedFields"] = ["Location"]
@@ -1518,13 +1662,13 @@ def search(ctx: typer.Context,
try:
with console.status(f"Retrieving search results..."):
r = requests.post(url,
- json=search_json_data,
- headers=headers(ctx)
- )
- except requests.exceptions.RequestException as e:
+ json=search_json_data,
+ headers=headers(ctx)
+ )
+ except requests.exceptions.RequestException as e:
raise SystemExit(e)
- try:
+ try:
# silly math for figuring out terminal size
width, height = os.get_terminal_size()
except:
@@ -1533,37 +1677,41 @@ def search(ctx: typer.Context,
if r.ok:
if "results" in r.json():
- count=len(r.json()["results"])
+ count = len(r.json()["results"])
if raw:
console.print(r.json())
elif not count:
error_console.print("No results found")
- raise typer.Exit(2) # non-zero exit status
+ raise typer.Exit(2) # non-zero exit status
elif id_only:
search_cli.display_search_results_idonly(r.json())
elif random_id:
search_cli.display_search_results_idonly_random(r.json())
elif output == output.fancy:
- search_cli.display_search_results_fancy(r.json(), show_kind=detail)
+ search_cli.display_search_results_fancy(
+ r.json(), show_kind=detail)
elif output == output.simple:
search_cli.display_search_results_simple(r.json())
elif output == output.excel:
with console.status(f"Saving search results..."):
- search_cli.display_search_results_excel(r.json(), show_kind=detail, excelfile=file.name)
+ search_cli.display_search_results_excel(
+ r.json(), show_kind=detail, excelfile=file.name)
#os.system(f"open -a'Microsoft Excel.app' {file.name}")
- #os.startfile(f"{file.name}")
+ # os.startfile(f"{file.name}")
if open_file:
search_cli.open_file(file.name)
elif output == output.tree:
error_console.print("Error: Not supported output type")
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
if with_cursor:
- console.print(f":warning-emoji: [bold] Cursor: [blue]{r.json()['cursor']}")
+ console.print(
+ f":warning-emoji: [bold] Cursor: [blue]{r.json()['cursor']}")
else:
error_console.print("Error:", r.text, r.status_code)
- raise typer.Exit(1) # non-zero exit status
+ raise typer.Exit(1) # non-zero exit status
+
if __name__ == "__main__":
cli()