Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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()
# Authenticate to CARTO account
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)
dataset_manager = DatasetManager(auth_client)
else:
logger.error('You need to provide valid credentials, run with -h parameter for details')
import sys
sys.exit(1)
# SQL wrapper
sql = SQLClient(APIKeyAuthClient(args.CARTO_BASE_URL, args.CARTO_API_KEY))
query = sql.send('EXPLAIN ANALYZE ' + args.queryUser)
for key, value in query.items():
if key == 'rows':
for itr in value:
logger.info(itr)
if key == 'time':
logger.info(str(key) + ': ' + str(value))
# 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)
else:
logger.error('You need to provide valid credentials, run with -h parameter for details')
import sys
sys.exit(1)
# get username from base_url
substring = re.search('https://(.+?).carto.com', args.CARTO_BASE_URL)
if substring:
username = substring.group(1)
# imports the file to CARTO
dataset_manager = DatasetManager(auth_client)
table = dataset_manager.create(args.url)
logger.info('Name of table: ' + str(table.name))
print('URL of dataset: \
https://{org}.carto.com/u/{username}/dataset/{data}'). \
format(org=args.organization,
username=username,
data=str(table.name))
# 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)
else:
logger.error('You need to provide valid credentials, run with -h parameter for details')
import sys
sys.exit(1)
# get username from base_url
substring = re.search('https://(.+?).carto.com', args.CARTO_BASE_URL)
if substring:
username = substring.group(1)
# Dataset manager
dataset_manager = DatasetManager(auth_client)
connection = json.loads(args.connection.replace("\\", ""))
logger.info(connection)
table = dataset_manager.create(None, None, connection=connection)
logger.info(
'Table imported: {table}'.format(table=table.name))
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()
# Authenticate to CARTO account
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)
dataset_manager = DatasetManager(auth_client)
else:
logger.error('You need to provide valid credentials, run with -h parameter for details')
import sys
sys.exit(1)
# SQL wrapper
sql = SQLClient(APIKeyAuthClient(args.CARTO_BASE_URL, args.CARTO_API_KEY))
# get username from base_url
substring = re.search('https://(.+?).carto.com', args.CARTO_BASE_URL)
if substring:
username = substring.group(1)
# check all table name of account
all_tables = []
' 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()
# Authenticate to CARTO account
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)
dataset_manager = DatasetManager(auth_client)
else:
logger.error('You need to provide valid credentials, run with -h parameter for details')
import sys
sys.exit(1)
# SQL wrapper
sql = SQLClient(APIKeyAuthClient(args.CARTO_BASE_URL, args.CARTO_API_KEY))
queries = "select pid, query from pg_stat_activity \
WHERE usename = current_user"
result = sql.send(queries)
for key, value in result.items():
if key == 'rows':
for itr in value:
logger.info(itr)
args.CARTO_BASE_URL, args.CARTO_API_KEY, args.organization)
else:
logger.error('You need to provide valid credentials, run with -h parameter for details')
import sys
sys.exit(1)
# get username from base_url
substring = re.search('https://(.+?).carto.com', args.CARTO_BASE_URL)
if substring:
username = substring.group(1)
# SQL wrapper
sql = SQLClient(APIKeyAuthClient(args.CARTO_BASE_URL, args.CARTO_API_KEY))
# Dataset manager
dataset_manager = DatasetManager(auth_client)
# define path of the files
path = os.getcwd()
file_folder = glob.glob(path + '/' + args.folder_name)
# import files from the path to CARTO
table_name = []
for i in file_folder:
table = dataset_manager.create(i)
logger.info(
'Table imported: {table}'.format(table=table.name))
table_name.append(table.name)
# define base table to insert all rows from other files
def tables(self):
"""List all tables in user's CARTO account
Returns:
:obj:`list` of :py:class:`Table `
"""
datasets = DatasetManager(self.auth_client).filter(
show_table_size_and_row_count='false',
show_table='false',
show_stats='false',
show_likes='false',
show_liked='false',
show_permission='false',
show_uses_builder_features='false',
show_synchronization='false',
load_totals='false')
return [Table.from_dataset(d) for d in datasets]
def _get_metadata(self, auth_client, table_name, retries=4, retry_wait_time=1):
ds_manager = DatasetManager(auth_client)
try:
return ds_manager.get(table_name)
except Exception as e:
if type(e).__name__ == 'NotFoundException' and retries > 0:
time.sleep(retry_wait_time)
self._get_metadata(auth_client=auth_client, table_name=table_name,
retries=retries-1, retry_wait_time=retry_wait_time*2)
else:
raise CartoException('We could not get the table metadata. '
'Please, try again in a few seconds or contact support for help')
def _get_privacy(self, table_name):
"""gets current privacy of a table"""
ds_manager = DatasetManager(self.auth_client)
try:
dataset = ds_manager.get(table_name)
return dataset.privacy.lower()
except NotFoundException:
return None