Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
size = IntegerField()
table_size = IntegerField()
map_id = CharField()
description = CharField()
geometry_types = CharField(many=True)
table_visualization = VisualizationField()
dependent_visualizations = None
non_dependent_visualizations = None
synchronization = SynchronizationField()
class Meta:
collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION)
name_field = "name"
class TableManager(Manager):
"""
Manager for the Table class.
.. warning:: Non-public API. It may change with no previous notice
"""
resource_class = Table
paginator_class = CartoPaginator
class DODataset(Resource):
"""
Represents a Data Observatory Datasets object in CARTO.
"""
dataset = CharField()
id = CharField()
project = CharField()
table = CharField()
class Meta:
collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION)
name_field = "id"
class DODatasetManager(Manager):
"""
Manager for the DODataset class.
"""
resource_class = DODataset
json_collection_attribute = "datasets"
paginator_class = CartoPaginator
super(FileImportJob, self).__init__(auth_client)
def run(self, **import_params):
"""
Actually creates the import job on the CARTO server
:param import_params: To be send to the Import API, see CARTO's docs on Import API for an updated list of accepted params
:return:
"""
if self.url:
import_params["url"] = self.url
super(FileImportJob, self).run(params=import_params, files=self.files)
self.id_field = "id"
class FileImportJobManager(Manager):
resource_class = FileImportJob
json_collection_attribute = "imports"
paginator_class = CartoPaginator
def filter(self):
"""
Get a filtered list of file imports
:return: A list of file imports, with only the id set (you need to refresh them if you want all the attributes to be filled in)
"""
response = self.send(self.get_collection_endpoint(), "get")
resource_ids = self.client.get_response_data(response, self.Meta.parse_json)[self.json_collection_attribute] if self.json_collection_attribute is not None else self.client.get_response_data(response, self.Meta.parse_json)
resources = []
for resource_id in resource_ids:
try:
def force_sync(self):
"""
Forces to sync the SyncTableJob
:return:
:raise: CartoException
"""
try:
self.send(self.get_resource_endpoint(), "put")
except Exception as e:
raise CartoException(e)
class SyncTableJobManager(Manager):
"""
Manager for the SyncTableJob class
"""
resource_class = SyncTableJob
json_collection_attribute = "synchronizations"
paginator_class = CartoPaginator
def create(self, url, interval, **kwargs):
"""
Create a sync table on the server
:param url: URL can be a pointer to a remote location or a path to a
local file
:param interval: Sync interval in seconds
:param kwargs: Attributes (field names and values) of the new resource
:type url: str
id = CharField()
estimated_delivery_days = FloatField()
subscription_list_price = FloatField()
tos = CharField()
tos_link = CharField()
licenses = CharField()
licenses_link = CharField()
rights = CharField()
type = CharField()
class Meta:
collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION)
name_field = "id"
class DOSubscriptionInfoManager(Manager):
"""
Manager for the DOSubscriptionInfo class.
"""
resource_class = DOSubscriptionInfo
json_collection_attribute = "subscription_info"
paginator_class = CartoPaginator
def get(self, id, type):
response = self.send(self.get_collection_endpoint(), "get", params={"id": id, "type": type})
try:
resource = self.resource_class(self.client)
except (ValueError, TypeError):
return None
else:
'This dataset contains dependent visualizations. ' +
'Delete them to be able to delete this dataset or use `force_delete` ' +
'to delete the dataset and the dependent visualizations.')
)
super(WarnResource, self).delete()
def force_delete(self):
super(WarnResource, self).delete()
class Meta:
collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION)
name_field = "name"
class DatasetManager(Manager):
"""
Manager for the Dataset class.
.. warning:: Non-public API. It may change with no previous notice
"""
resource_class = Dataset
json_collection_attribute = "visualizations"
paginator_class = CartoPaginator
def send(self, url, http_method, **client_args):
"""
Sends an API request, taking into account that datasets are part of
the visualization endpoint.
:param url: Endpoint URL
:param http_method: The method used to make the request to the API
self.send(endpoint, "POST")
except Exception as e:
raise CartoException(e)
def save(self):
pass
def refresh(self):
pass
def delete(self):
pass
class OauthAppManager(Manager):
"""
Manager for the OauthApp class.
"""
resource_class = OauthApp
json_collection_attribute = "result"
paginator_class = CartoPaginator
def create(self, name, redirect_uris, icon_url):
"""
Creates an OauthApp.
:param name: The OAuth app name
:param redirect_uris: An array of URIs for authorize callback.
:param icon_url: A URL with a squared icon for the Oauth app.
:type name: str
.. warning:: Non-public API. It may change with no previous notice
"""
access_token = CharField()
bq_public_project = CharField()
gcp_execution_project = CharField()
bq_project = CharField()
bq_dataset = CharField()
gcs_bucket = CharField()
instant_licensing = BooleanField()
class Meta:
collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION)
name_field = "access_token"
class DoTokenManager(Manager):
"""
Manager for the DoToken class.
.. warning:: Non-public API. It may change with no previous notice
"""
resource_class = DoToken
json_collection_attribute = None
paginator_class = CartoPaginator
def get(self):
return super(DoTokenManager, self).get('token')
"""
Represents a Data Observatory Subscriptions in CARTO.
"""
dataset = CharField()
id = CharField()
project = CharField()
table = CharField()
type = CharField()
class Meta:
collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION)
name_field = "id"
class DOSubscriptionManager(Manager):
"""
Manager for the DOSubscription class.
"""
resource_class = DOSubscription
json_collection_attribute = "subscriptions"
paginator_class = CartoPaginator
class DOCreatedSubscription(Resource):
"""
Represents a Data Observatory Subscriptions in CARTO.
"""
id = CharField()
estimated_delivery_days = FloatField()
id = CharField()
estimated_delivery_days = FloatField()
subscription_list_price = FloatField()
tos = CharField()
tos_link = CharField()
licenses = CharField()
licenses_link = CharField()
rights = CharField()
type = CharField()
class Meta:
collection_endpoint = API_ENDPOINT.format(api_version=API_VERSION)
name_field = "id"
class DOSubscriptionCreationManager(Manager):
"""
Manager for the DOSubscription class.
"""
resource_class = DOCreatedSubscription
json_collection_attribute = "subscriptions"
paginator_class = CartoPaginator