External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-11 01:57:35 -03:30
Code Issues Packages Projects Releases Wiki Activity

Move the export logic onto methods on the ApiV2 class

Browse Source 
making it easier to invoke programmatically.
This commit is contained in:
Jeff Bradberry 2020-03-20 14:02:12 -04:00
parent 19c92a7055
commit e4146e9bc7
2 changed files with 119 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
awxkit/awxkit/api/pages/api.py
View File

@ -3,6 +3,60 @@ from . import base
from . import page
EXPORTABLE_RESOURCES = [
'users',
'organizations',
'teams',
'credential_types',
'credentials',
'notification_templates',
# 'projects',
# 'inventory',
# 'job_templates',
# 'workflow_job_templates',
]
NATURAL_KEYS = {
'user': ('username',),
'organization': ('name',),
'team': ('organization', 'name'),
'credential_type': ('name', 'kind'),
'credential': ('organization', 'name', 'credential_type'),
'notification_template': ('organization', 'name'),
'project': ('organization', 'name'),
'inventory': ('organization', 'name'),
'job_template': ('organization', 'name'),
'workflow_job_template': ('organization', 'name'),
# related resources
'role': ('name', ':content_object'),
}
def get_natural_key(page):
natural_key = {'type': page['type']}
lookup = NATURAL_KEYS.get(page['type'], ())
for key in lookup or ():
if key.startswith(':'):
# treat it like a special-case related object
related_objs = [
related for name, related in page.related.items()
if name not in ('users', 'teams')
]
if related_objs:
natural_key[key[1:]] = get_natural_key(related_objs[0].get())
elif key in page.related:
natural_key[key] = get_natural_key(page.related[key].get())
elif key in page:
natural_key[key] = page[key]
if not natural_key:
return None
return natural_key
class Api(base.Base):
pass
@ -13,7 +67,67 @@ page.register_page(resources.api, Api)
class ApiV2(base.Base):
pass
def _get_options(self, endpoint):
return endpoint.options().json['actions']['POST']
def _serialize_asset(self, asset, options):
# Drop any (credential_type) assets that are being managed by the Tower instance.
if asset.json.get('managed_by_tower'):
return None
fields = {
key: asset[key] for key in options
if key in asset.json and key not in asset.related
}
fields['natural_key'] = get_natural_key(asset)
fk_fields = {
key: get_natural_key(asset.related[key].get()) for key in options
if key in asset.related
}
related = {}
for k, related_endpoint in asset.related.items():
if k != 'roles':
continue
data = related_endpoint.get(all_pages=True)
if 'results' in data:
related[k] = [get_natural_key(x) for x in data.results]
related_fields = {'related': related} if related else {}
fields.update(fk_fields)
fields.update(related_fields)
return fields
def _get_assets(self, resource, value):
endpoint = getattr(self, resource)
if value:
from awxkit.cli.options import pk_or_name
pk = pk_or_name(self, resource, value)
results = endpoint.get(id=pk).results
else:
results = endpoint.get(all_pages=True).results
options = self._get_options(endpoint)
assets = (self._serialize_asset(asset, options) for asset in results)
return [asset for asset in assets if asset is not None]
def export_assets(self, **kwargs):
# If no resource kwargs are explicitly used, export everything.
all_resources = all(kwargs.get(resource) is None for resource in EXPORTABLE_RESOURCES)
data = {}
for resource in EXPORTABLE_RESOURCES:
value = kwargs.get(resource)
if all_resources or value is not None:
data[resource] = self._get_assets(resource, value)
return data
def import_assets(self):
pass
page.register_page(resources.v2, ApiV2)

116
awxkit/awxkit/cli/resource.py
View File

@ -8,6 +8,7 @@ import awxkit.exceptions as exc
from awxkit.utils import to_str
from awxkit.api.mixins import has_create
from awxkit.api.pages import Page
from awxkit.api.pages.api import EXPORTABLE_RESOURCES, get_natural_key
from awxkit.cli.format import FORMATTERS, format_response, add_authentication_arguments
from awxkit.cli.utils import CustomRegistryMeta, cprint
@ -47,58 +48,6 @@ DEPRECATED_RESOURCES_REVERSE = dict(
(v, k) for k, v in DEPRECATED_RESOURCES.items()
)
EXPORTABLE_RESOURCES = [
'users',
'organizations',
'teams',
'credential_types',
'credentials',
'notification_templates',
# 'projects',
# 'inventory',
# 'job_templates',
# 'workflow_job_templates',
]
NATURAL_KEYS = {
'user': ('username',),
'organization': ('name',),
'team': ('organization', 'name'),
'credential_type': ('name', 'kind'),
'credential': ('organization', 'name', 'credential_type'),
'notification_template': ('organization', 'name'),
'project': ('organization', 'name'),
'inventory': ('organization', 'name'),
'job_template': ('organization', 'name'),
'workflow_job_template': ('organization', 'name'),
# related resources
'role': ('name', ':content_object'),
}
def get_natural_key(page):
natural_key = {'type': page['type']}
lookup = NATURAL_KEYS.get(page['type'], ())
for key in lookup or ():
if key.startswith(':'):
# treat it like a special-case related object
related_objs = [
related for name, related in page.related.items()
if name not in ('users', 'teams')
]
if related_objs:
natural_key[key[1:]] = get_natural_key(related_objs[0].get())
elif key in page.related:
natural_key[key] = get_natural_key(page.related[key].get())
elif key in page:
natural_key[key] = page[key]
if not natural_key:
return None
return natural_key
class CustomCommand(metaclass=CustomRegistryMeta):
"""Base class for implementing custom commands.
@ -317,53 +266,6 @@ class Export(CustomCommand):
# 3) the resource flag is used with an argument, and the attr will be that argument's value
resources.add_argument('--{}'.format(resource), nargs='?', const='')
def get_options(self, endpoint):
return endpoint.options().json['actions']['POST']
def get_assets(self, resource, value):
endpoint = getattr(self.v2, resource)
if value:
from .options import pk_or_name
pk = pk_or_name(self.v2, resource, value)
results = endpoint.get(id=pk).results
else:
results = endpoint.get(all_pages=True).results
options = self.get_options(endpoint)
assets = (self.serialize_asset(asset, options) for asset in results)
return [asset for asset in assets if asset is not None]
def serialize_asset(self, asset, options):
# Drop any (credential_type) assets that are being managed by the Tower instance.
if asset.json.get('managed_by_tower'):
return None
fields = {
key: asset[key] for key in options
if key in asset.json and key not in asset.related
}
fields['natural_key'] = get_natural_key(asset)
fk_fields = {
key: get_natural_key(asset.related[key].get()) for key in options
if key in asset.related
}
related = {}
for k, related_endpoint in asset.related.items():
if k != 'roles':
continue
data = related_endpoint.get(all_pages=True)
if 'results' in data:
related[k] = [get_natural_key(x) for x in data.results]
related_fields = {'related': related} if related else {}
fields.update(fk_fields)
fields.update(related_fields)
return fields
def handle(self, client, parser):
self.extend_parser(parser)
@ -371,21 +273,11 @@ class Export(CustomCommand):
parser.print_help()
raise SystemExit()
client.authenticate()
parsed = parser.parse_known_args()[0]
kwargs = {resource: getattr(parsed, resource, None) for resource in EXPORTABLE_RESOURCES}
# If no resource flags are explicitly used, export everything.
all_resources = all(getattr(parsed, resource, None) is None for resource in EXPORTABLE_RESOURCES)
self.v2 = client.v2
data = {}
for resource in EXPORTABLE_RESOURCES:
value = getattr(parsed, resource, None)
if all_resources or value is not None:
data[resource] = self.get_assets(resource, value)
return data
client.authenticate()
return client.v2.export_assets(**kwargs)
def parse_resource(client, skip_deprecated=False):