Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def sync_table_manager(api_key_auth_client):
"""
Returns a sync table manager that can be reused in tests
:param api_key_auth_client: Fixture that provides a valid APIKeyAuthClient
object
:return: SyncTableJobManager instance
"""
return SyncTableJobManager(api_key_auth_client)
help='Set the base URL. For example:' +
' https://username.carto.com/ ' +
'(defaults to env variable CARTO_API_URL)')
parser.add_argument('--api_key', dest='CARTO_API_KEY',
default=os.environ['CARTO_API_KEY'] if 'CARTO_API_KEY' in os.environ else '',
help='Api key of the account' +
' (defaults to env variable CARTO_API_KEY)')
args = parser.parse_args()
# Set authentification to CARTO
if args.CARTO_BASE_URL and args.CARTO_API_KEY and args.organization:
auth_client = APIKeyAuthClient(
args.CARTO_BASE_URL, args.CARTO_API_KEY, args.organization)
syncTableManager = SyncTableJobManager(auth_client)
syncTable = syncTableManager.create(args.url, args.sync_time)
else:
logger.error('You need to provide valid credentials, run with -h parameter for details')
import sys
sys.exit(1)
# return the id of the sync
logging.debug((syncTable.get_id()))
while(syncTable.state != 'success'):
time.sleep(5)
syncTable.refresh()
logging.debug(syncTable.state)
if (syncTable.state == 'failure'):
logging.warn('The error code is: ' + str(syncTable.error_code))
logging.warn('The error message is: ' + str(syncTable.error_message))
If not None, CARTO will try to set up a sync table
against the (remote) URL
:param import_args: Arguments to be sent to the import job when run
:type archive: str
:type interval: int
:type import_args: kwargs
:return: New dataset object
:rtype: Dataset
:raise: CartoException
"""
archive = archive.lower() if hasattr(archive, "lower") else archive
if self.is_sync_table(archive, interval, **import_args):
manager = SyncTableJobManager(self.client)
else:
manager = FileImportJobManager(self.client)
import_job = manager.create(archive) if interval is None \
else manager.create(archive, interval)
import_job.run(**import_args)
if import_job.get_id() is None:
raise CartoException(_("Import API returned corrupt job details \
when creating dataset"))
import_job.refresh()
count = 0
while import_job.state in ("enqueued", "queued", "pending", "uploading",
"unpacking", "importing", "guessing") \
manager = FileImportJobManager(self.client)
import_job = manager.create(archive) if interval is None \
else manager.create(archive, interval)
import_job.run(**import_args)
if import_job.get_id() is None:
raise CartoException(_("Import API returned corrupt job details \
when creating dataset"))
import_job.refresh()
count = 0
while import_job.state in ("enqueued", "queued", "pending", "uploading",
"unpacking", "importing", "guessing") \
or (isinstance(manager, SyncTableJobManager)
and import_job.state == "created"):
if count >= MAX_NUMBER_OF_RETRIES:
raise CartoException(_("Maximum number of retries exceeded \
when polling the import API for \
dataset creation"))
time.sleep(INTERVAL_BETWEEN_RETRIES_S)
import_job.refresh()
count += 1
if import_job.state == "failure":
raise CartoException(_("Dataset creation was not successful \
because of failed import (error: {error}")
.format(error=json.dumps(
import_job.get_error_text)))
if (import_job.state != "complete" and import_job.state != "created"