Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
def __init__(self, file_name, client, api_version='v1', **kwargs):
"""
:param file_name: File name (paths are supported)
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param kwargs: Sent to parent class
:return:
"""
self.files = {'file': open(file_name, 'rb')}
self.api_url = IMPORT_API_FILE_URL.format(api_version=api_version)
super(FileImport, self).__init__(client, **kwargs)
class URLImport(ImportJob):
"""
This class provides support for uploading and importing remote files into CartoDB
Sync tables are created if the interval param is used
"""
def __init__(self, url, client, api_version='v1', interval=None, **kwargs):
"""
:param url: Remote URL for the file
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param interval: Number of seconds between update intervals (>=900). If none, URL won't be sync'ed
:param kwargs: Sent to parent class
:return:
"""
self.url = url
self.interval = interval
self.api_url = IMPORT_API_SYNC_TABLE_URL.format(api_version=api_version)
"""
:param url: Remote URL for the file
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param interval: Number of seconds between update intervals (>=900). If none, URL won't be sync'ed
:param kwargs: Sent to parent class
:return:
"""
self.url = url
self.interval = interval
self.api_url = IMPORT_API_SYNC_TABLE_URL.format(api_version=api_version)
super(URLImport, self).__init__(client, **kwargs)
class ExportJob(ImportJob):
def __init__(self, client, visualization_id, api_key, api_version='v3', **kwargs):
self.viz_id = visualization_id
self.api_key = client.api_key
self.api_url = IMPORT_API_EXPORT_URL.format(api_version=api_version)
super(ExportJob, self).__init__(client, **kwargs)
def run(self, **import_params):
import_params["visualization_id"] = self.viz_id
self.send(self.api_url, url_params=import_params, client_params={"http_method": "POST"})
self.send(self.api_url + self.id, client_params = {"http_method": "GET"})
def update(self):
if self.id is None:
raise CartoException("Export job needs to be run or retrieved first!")
def get(self, id=None, ids_only=False):
"""
Get one import job or a list with all the current (pending) import jobs
:param id: If set, only this job will be retrieved. This works no matter the state of the job
:param ids_only: If True, a list of IDs is returned; if False, a list of ImportJob objects is returned
:return: An import job, a list of import job IDs or a list of import jobs
"""
if id is not None:
resp = self.client.send("%s/%s" % (self.api_url, id))
response_data = self.client.get_response_data(resp, True)
return ImportJob(self.client, **response_data)
else:
imports = []
resp = self.client.send(self.api_url)
response_data = self.client.get_response_data(resp, True)
if response_data.get("success", False) is not False:
for import_job_id in response_data["imports"]:
if ids_only is True:
imports.append(import_job_id)
else:
imports.append(self.get(import_job_id))
return imports
self.send(self.api_url, client_params={"json": import_params, "http_method": "POST"})
elif self.files is not None: # File import
self.send(self.api_url, url_params=import_params, client_params={"http_method": "POST", "files": self.files})
def update(self):
"""
Updates the information of the import job against the CartoDB server
:return:
"""
if self.id is None:
raise CartoException("Import job needs to be run or retrieved first!")
self.send("%s/%s" % (self.api_url, self.id))
class FileImport(ImportJob):
"""
This class provides support for uploading and importing local files into CartoDB
"""
def __init__(self, file_name, client, api_version='v1', **kwargs):
"""
:param file_name: File name (paths are supported)
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param kwargs: Sent to parent class
:return:
"""
self.files = {'file': open(file_name, 'rb')}
self.api_url = IMPORT_API_FILE_URL.format(api_version=api_version)
super(FileImport, self).__init__(client, **kwargs)
def get(self, id=None, ids_only=False):
"""
Get one import job or a list with all the current (pending) import jobs
:param id: If set, only this job will be retrieved. This works no matter the state of the job
:param ids_only: If True, a list of IDs is returned; if False, a list of ImportJob objects is returned
:return: An import job, a list of import job IDs or a list of import jobs
"""
if id is not None:
resp = self.client.send("%s/%s" % (self.api_url, id))
response_data = self.client.get_response_data(resp, True)
return ImportJob(self.client, **response_data)
else:
imports = []
resp = self.client.send(self.api_url)
response_data = self.client.get_response_data(resp, True)
if response_data.get("success", False) is not False:
for import_job_id in response_data["imports"]:
if ids_only is True:
imports.append(import_job_id)
else:
imports.append(self.get(import_job_id))
return imports
"""
:param url: Remote URL for the file
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param interval: Number of seconds between update intervals (>=900). If none, URL won't be sync'ed
:param kwargs: Sent to parent class
:return:
"""
self.url = url
self.interval = interval
self.api_url = IMPORT_API_SYNC_TABLE_URL.format(api_version=api_version)
super(URLImport, self).__init__(client, **kwargs)
class ExportJob(ImportJob):
"""
Equivalent to a carto export in CARTO.
Allows a carto export to be created using a visualization in the user's CARTO account
"""
def __init__(self, client, visualization_id, api_version='v3', **kwargs):
"""
:param client: Client to make authorized requests
:param visualization_id: The id of the map that will be exported
:return:
"""
self.viz_id = visualization_id
self.api_url = IMPORT_API_EXPORT_URL.format(api_version=api_version)
super(ExportJob, self).__init__(client, **kwargs)
def run(self, **import_params):
self.send(self.api_url, client_params={"json": import_params, "http_method": "POST"})
elif self.files is not None: # File import
self.send(self.api_url, url_params=import_params, client_params={"http_method": "POST", "files": self.files})
def update(self):
"""
Updates the information of the import job against the CARTO server
:return:
"""
if self.id is None:
raise CartoException("Import job needs to be run or retrieved first!")
self.send("%s/%s" % (self.api_url, self.id))
class FileImport(ImportJob):
"""
This class provides support for uploading and importing local files into CARTO
"""
def __init__(self, file_name, client, api_version='v1', **kwargs):
"""
:param file_name: File name (paths are supported)
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param kwargs: Sent to parent class
:return:
"""
self.files = {'file': open(file_name, 'rb')}
self.api_url = IMPORT_API_FILE_URL.format(api_version=api_version)
super(FileImport, self).__init__(client, **kwargs)
"""
def __init__(self, file_name, client, api_version='v1', **kwargs):
"""
:param file_name: File name (paths are supported)
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param kwargs: Sent to parent class
:return:
"""
self.files = {'file': open(file_name, 'rb')}
self.api_url = IMPORT_API_FILE_URL.format(api_version=api_version)
super(FileImport, self).__init__(client, **kwargs)
class URLImport(ImportJob):
"""
This class provides support for uploading and importing remote files into CARTO
Sync tables are created if the interval param is used
"""
def __init__(self, url, client, api_version='v1', interval=None, **kwargs):
"""
:param url: Remote URL for the file
:param client: Client to make authorized requests (currently only APIKeyAuthClient is supported)
:param api_version: Only 'v1' is currently supported
:param interval: Number of seconds between update intervals (>=900). If none, URL won't be sync'ed
:param kwargs: Sent to parent class
:return:
"""
self.url = url
self.interval = interval
self.api_url = IMPORT_API_SYNC_TABLE_URL.format(api_version=api_version)