Skip to content

Instantly share code, notes, and snippets.

@dgraziotin

dgraziotin/generate_clean_api_client.py

Last active December 15, 2015 08:08
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dgraziotin/5228154 to your computer and use it in GitHub Desktop.
Save dgraziotin/5228154 to your computer and use it in GitHub Desktop.
Download ZIP
Automatically generate a bare Python API client for CKAN v2.0
Raw
generate_clean_api_client.py
import os
import inspect
import shutil
import docutils.core
import ckan.logic.action.get as getactions
import settings_clean_api_client as settings
def extract_func_params_from_doc(docstring):
params = []
# thank you, http://stackoverflow.com/a/11315548
doctree = docutils.core.publish_doctree(docstring).asdom()
fields = doctree.getElementsByTagName('field_name')
for field in fields:
field_value = field.firstChild.nodeValue
if 'param' in field_value:
params.append(str(field_value.split()[1]).replace('.', '_'))
return params
def func_params_as_string(param_list):
param_list = [param+"=''" for param in param_list]
param_list = ", ".join(param_list )
param_list = ', ' + param_list
return param_list
def am_i_dangerous():
if os.path.exists(settings.API_BASE_DIR) or os.path.exists(settings.API_TESTS_DIR):
return True
return False
if __name__=="__main__":
if am_i_dangerous():
print "WARNING"
print "This script _overwrites_ the content of the api client and tests folders."
user_continue = raw_input("Would you like to continue? Y/N: ").upper()
if user_continue != 'Y':
exit(0)
print "Generating the API client functions and tests"
try:
shutil.rmtree(settings.API_BASE_DIR)
shutil.rmtree(settings.API_TESTS_DIR)
except OSError:
pass #yeah, yeah..
os.makedirs(settings.API_BASE_DIR)
os.makedirs(settings.API_TESTS_DIR)
concerns = settings.API_CONCERNS
for concern_name in concerns:
concern_filename = concern_name + ".py"
test_concern_filename = "test_" + concern_filename
with open(os.path.join(settings.API_BASE_DIR,concern_filename), 'a') as concern_module_file, open(os.path.join(settings.API_TESTS_DIR,test_concern_filename), 'a') as test_concern_module_file:
concern_module_file.write(settings.API_MODULE_TEMPLATE)
test_concern_module_file.write(settings.API_TEST_MODULE_TEMPLATE)
concern_function_names = concerns[concern_name]
for function_name in concern_function_names:
function_obj = getattr(getactions, function_name)
function_doc = function_obj.__doc__
param_list = None
param_list_str = ''
if function_doc:
param_list = extract_func_params_from_doc(function_doc)
if param_list:
param_list_str = func_params_as_string(param_list)
first_param_doc = function_doc.find(':param')
function_description_first = function_doc[:first_param_doc]
function_description_last = function_doc[first_param_doc:]
else:
function_description_first = '<TODO: DESCRIPTION> '
function_description_last = '<TODO: PARAMETERS>'
param_list = ''
concern_module_file.write(settings.API_FUNC_TEMPLATE % (
function_name, param_list_str, function_description_first, function_description_last, function_name
))
test_concern_module_file.write(settings.API_TEST_FUNC_TEMPLATE % (
function_name, function_name, param_list_str
))
# generate __init__.py for "get" package, to act as a huge "get" module
with open(os.path.join(settings.API_BASE_DIR,'__init__.py'), 'a') as init_file:
init_file.write('from ' + concern_name + ' import *\n')
# generate __init__.py for "get" package, to act as a huge "get" module
with open(os.path.join(settings.API_TESTS_DIR,'__init__.py'), 'a') as init_file:
init_file.write('from test_' + concern_name + ' import *\n')
print "Finished."
print "Your CKAN API v3.0 functions are in " + settings.API_BASE_DIR
print "The auto-generated (non working) tests are in " + settings.API_TESTS_DIR
print "Please, add the __init__.py files in the sub-directories if needed."
Raw
settings_clean_api_client.py
import os
API_BASE_DIR = os.path.join('libckan','logic','action','get')
API_TESTS_DIR = os.path.join('tests','logic','action','get')
"""
obtained with:
import ckan.logic.action.get as pymodule
import inspect
def get_public_functions(pymodule):
return [f for f in inspect.getmembers(pymodule, inspect.isfunction)
if not f[0].startswith("_")]
"""
API_CONCERNS = {
'activity' : [
'dashboard_activity_list',
'dashboard_activity_list_html',
'dashboard_new_activities_count',
'activity_detail_list',
'group_activity_list',
'group_activity_list_html',
'organization_activity_list',
'organization_activity_list_html',
'package_activity_list',
'package_activity_list_html',
'user_activity_list',
'user_activity_list_html',
'recently_changed_packages_activity_list',
'recently_changed_packages_activity_list_html',
],
'follow' : [
'group_followee_count',
'group_followee_list',
'group_follower_count',
'group_follower_list',
'dataset_followee_count',
'dataset_followee_list',
'dataset_follower_count',
'dataset_follower_list',
'followee_count',
'followee_list',
'user_followee_count',
'user_followee_list',
'user_follower_count',
'user_follower_list',
'am_following_dataset',
'am_following_group',
'am_following_user',
],
'group': [
'group_list',
'group_list_authz',
'group_show',
'group_show_rest',
],
'organization' : [
'organization_list',
'organization_list_for_user',
'organization_show',
],
'package' : [
'package_autocomplete',
'package_list',
'package_relationships_list',
'package_search',
'package_show',
'package_show_rest',
'current_package_list_with_resources',
'group_package_show',
],
'user': [
'user_autocomplete',
'user_list',
'user_show',
'get_site_user',
'member_list',
],
'related' : [
'related_list',
'related_show',
],
'resource' : [
'resource_search',
'resource_show',
],
'revision' : [
'revision_list',
'revision_show',
'package_revision_list',
'group_revision_list',
],
'roles' : [
'roles_show',
'member_roles_list',
],
'status' : [
'status_show',
'resource_status_show',
'task_status_show',
],
'tag' : [
'tag_autocomplete',
'tag_list',
'tag_search',
'tag_show',
'tag_show_rest',
],
'term' : [
'term_translation_show',
'vocabulary_list',
'vocabulary_show',
'format_autocomplete',
],
'misc' : [
'licence_list',
'site_read',
]
}
API_MODULE_TEMPLATE = (
"""import libckan.model.client as client
import libckan.model.exceptions as exceptions
"""
)
API_FUNC_TEMPLATE = (
"""def %s(client=client.Client()%s):
\"\"\"
%s
:param client: the CKAN Client.
Default: an instance of libckan.model.client.Client
:type client: libckan.model.client.Client
%s
:returns: the dictionary returned by the CKAN API,
with the keys "help","result", and "success".
"results" is a list of packages (dict).
:return: dict
Raises: :class:`libckan.model.exceptions.CKANError`:
An error occurred accessing CKAN API
\"\"\"
args = client.sanitize_params(locals())
resp = client.request(action='%s', data=args)
if not resp['success']:
raise exceptions.CKANError(resp.error)
return resp
"""
)
API_TEST_FUNC_TEMPLATE = (
"""def test_%s():
results = get.%s(client=client.Client()%s)
assert results['success'] is True
assert results['result'] is True
"""
)
API_TEST_MODULE_TEMPLATE = (
"""import nose.tools
import libckan.logic.action.get as get
import libckan.model.client as client
import libckan.model.exceptions as exceptions
"""
)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment