Skip to content

[WIP] Add more optional field arguments to create user command #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ShaiahWren wants to merge 4 commits into from
Closed

[WIP] Add more optional field arguments to create user command #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fe50fa1
add assorted kwargs
ShaiahWren Dec 16, 2021
9592155
rewriting main.py() argparsers
ShaiahWren Dec 17, 2021
8896d2c
First crack at removing all the unneeded functionality
Dec 17, 2021
b4336f0
Merge remote-tracking branch 'upstream/refactor-command.py' into add-…
ShaiahWren Dec 17, 2021
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
233 changes: 60 additions & 173 deletions galaxykit/command.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@


def print_unknown_error(args):
print(f"Unknown {args.kind} operation '{args.operation}'")
print(f"Unknown {args.object} operation '{args.operation}'")
sys.exit(EXIT_UNKNOWN_ERROR)


Expand All @@ -39,17 +39,62 @@ def report_error(resp):
f"API Failure: HTTP {error['status']} {error['code']}; {error['title']} ({error['detail']})"
)

def user():

return None

def namespace():
return None

def collection():
return None

def group():
return None

def remote_registry():
return None

def execution_environment():
return None


def user(parent_parser, client):
parser = argparse.ArgumentParser(parent=parent_parser)
args = parser.parse_args(parent_parser.rest)
return None


def namespace(parent_parser, client):
return None


def collection(parent_parser, client):
return None


def group(parent_parser, client):
return None


def registry(parent_parser, client):
return None


def container(parent_parser, client):
return None


def main():
parser = argparse.ArgumentParser()
parser.add_argument(
"kind",
"object",
type=str,
action="store",
help="Kind of API content to operate against (user, group, namespace)",
help="Type of API content to operate against (user, group, namespace)",
)
parser.add_argument("operation", type=str, action="store")
parser.add_argument("rest", type=str, action="store", nargs="*")
parser.add_argument("remaining", type=str, action="store", nargs="*")
parser.add_argument("-i", "--ignore", default=False, action="store_true")
parser.add_argument("-u", "--username", type=str, action="store")
parser.add_argument("-p", "--password", type=str, action="store")
Expand All @@ -74,178 +119,20 @@ def main():
client = GalaxyClient(
args.server, (args.username, args.password), https_verify=https_verify
)
resp = None

try:
if args.kind == "user":
if args.operation == "list":
resp = users.get_user_list(client)
print(format_list(resp["data"], "username"))
elif args.operation == "create":
username, password = args.rest
created, resp = users.get_or_create_user(
client, username, password, None
)
if created:
print("Created user", username)
else:
print(f"User {username} already existed")
elif args.operation == "delete":
(username,) = args.rest
try:
resp = users.delete_user(client, username)
except ValueError as e:
if not args.ignore:
print(e)
sys.exit(EXIT_NOT_FOUND)
elif args.operation == "group":
subop, *subopargs = args.rest
if subop == "add":
username, groupname = subopargs
user_data = users.get_user(client, username)
group_id = groups.get_group_id(client, groupname)
user_data["groups"].append(
{
"id": group_id,
"name": groupname,
"pulp_href": f"/pulp/api/v3/groups/{group_id}",
}
)
resp = users.update_user(client, user_data)
else:
print_unknown_error(args)

elif args.kind == "group":
if args.operation == "list":
resp = groups.get_group_list(client)
print(format_list(resp["data"], "name"))
elif args.operation == "create":
(name,) = args.rest
resp = groups.create_group(client, name)
elif args.operation == "delete":
(name,) = args.rest
try:
resp = groups.delete_group(client, name)
except ValueError as e:
if not args.ignore:
print(e)
sys.exit(EXIT_NOT_FOUND)
elif args.operation == "perm":
subop, *subopargs = args.rest
if subop == "list":
(groupname,) = subopargs
resp = groups.get_permissions(client, groupname)
print(format_list(resp["data"], "permission"))
elif subop == "add":
groupname, perm = subopargs
perms = [
p["permission"]
for p in groups.get_permissions(client, groupname)["data"]
]
perms = list(set(perms) | set([perm]))
resp = groups.set_permissions(client, groupname, perms)
elif subop == "remove":
groupname, perm = subopargs
resp = groups.delete_permission(client, groupname, perm)
else:
print(f"Unknown group perm operation '{subop}'")
sys.exit(EXIT_UNKNOWN_ERROR)
else:
print_unknown_error(args)

elif args.kind == "namespace":
if args.operation == "get":
(name,) = args.rest
print(json.dumps(namespaces.get_namespace(client, name)))
elif args.operation == "list-collections":
(name,) = args.rest
print(json.dumps(namespaces.get_namespace_collections(client, name)))
elif args.operation == "create":
if len(args.rest) == 2:
name, group = args.rest
else:
(name,) = args.rest
group = None
resp = namespaces.create_namespace(client, name, group)
elif args.operation == "delete":
raise NotImplementedError
elif args.operation == "groups":
raise NotImplementedError
elif args.operation == "addgroup":
name, group = args.rest
resp = namespaces.add_group(client, name, group)
elif args.operation == "removegroup":
name, group = args.rest
resp = namespaces.remove_group(client, name, group)
elif args.operation == "addgroupperm":
raise NotImplementedError
elif args.operation == "removegroupperm":
raise NotImplementedError
else:
print_unknown_error(args)

elif args.kind == "container":
if args.operation == "readme":
if len(args.rest) == 1:
(container,) = args.rest
resp = containers.get_readme(client, container)
print(resp["text"])
elif len(args.rest) == 2:
container, readme = args.rest
resp = containers.set_readme(client, container, readme)
else:
print("container readme takes either 1 or 2 parameters.")
sys.exit(EXIT_UNKNOWN_ERROR)
else:
print_unknown_error(args)

elif args.kind == "collection":
if args.operation == "upload":
if len(args.rest) == 0:
(namespace, collection_name) = (client.username, None)
else:
(namespace, collection_name) = args.rest

resp = namespaces.create_namespace(client, namespace, None)
artifact = collections.upload_test_collection(
client, namespace=namespace, collection_name=collection_name
)
print(json.dumps(artifact))
elif args.operation == "move":
if len(args.rest) == 2:
(namespace, collection_name) = args.rest
# defaults to version = 1.0.0, source = staging, destination = published
collections.move_collection(client, namespace, collection_name)
else:
(
namespace,
collection_name,
version,
source,
destination,
) = args.rest
collections.move_collection(
client, namespace, collection_name, version, source, destination
)
else:
print_unknown_error(args)

elif args.kind == "url":
if args.operation == "get":
(url,) = args.rest
print(json.dumps(client.get(url)))
elif args.operation == "post":
raise NotImplementedError
else:
print_unknown_error(args)

funs = {
"user": user,
"group": group,
"namespace": namespace,
"container": container,
"collection": collection,
"url": url,
}
if args.object in funs:
funs[args.object](args.rest, client)
else:
print(f"Unknown resource type '{args.kind}'")
print(f"Unknown resource type '{args.object}'")
sys.exit(EXIT_UNKNOWN_ERROR)

if resp and not ignore:
report_error(resp)

except GalaxyClientError:
if not ignore:
raise
Loading