Skip to content

Queries

Class QueriesService

class QueriesService(FavoriteMixin, CommonMixin, TagsMixin, PublishMixin,
                     NameMixin, PrintMixin, PaginationMixin)

[view_source]

refresh

def refresh(query_id: int) -> Dict

[view_source]

Refresh a query

fork

def fork(query_id: int, new_name: str = None) -> Dict

[view_source]

Fork a query with a new name if provided

scheduled

def scheduled() -> List[Dict]

[view_source]

Get all scheduled queries

update_visualization

def update_visualization(vis_id: int, data: Dict) -> Dict

[view_source]

Update a visualization with data

duplicate_query_table

def duplicate_query_table(*,
                          query: Dict,
                          new_name: str = None,
                          table_map: Dict[str, str],
                          tags: List[str],
                          new_data_source_id: Optional[int] = None,
                          publish: bool = True) -> Dict

[view_source]

Duplicate query with source tables replaced according to table_map.

Arguments:

  • query(Dict) - query data object
  • new_name(str) - new query name (optional)
  • table_map(Dict) - mapping of old to new tables
  • tags(List) - tags to add to new query
  • new_data_source_id(int) - new data source id (optional)
  • publish(bool) - whether to publish the new query

Returns:

  • Dict - new query data object

Example:

query = queries.get(query_id) table_map = { "old_table": "new_table", "old_table2": "new_table2", } tags = ["tag1", "tag2"] new_query = queries.duplicate_query_table( query=query, new_name="new_query_name", table_map=table_map, new_data_source_id=rd.data_sources.get_by_name("new_data_source").id, tags=tags, )

Class FavoriteMixin

class FavoriteMixin()

[view_source]

Mixin for favoriteable services

favorited

def favorited() -> List[Dict]

[view_source]

Get all favorited objects

favorite

def favorite(id: int) -> Dict

[view_source]

Favorite an object

unfavorite

def unfavorite(id: int) -> Dict

[view_source]

Unfavorite an object

Class CommonMixin

class CommonMixin()

[view_source]

Mixin with common methods for services.

get

def get(id: int) -> Dict

[view_source]

Fetch one by ID

get_all

def get_all() -> List[Dict]

[view_source]

fetch all objects.

update

def update(id: int, data: Dict) -> Dict

[view_source]

Update by ID

create

def create(data: Dict) -> Dict

[view_source]

Create a new object with data

delete

def delete(id: int) -> None

[view_source]

Delete by ID

Class TagsMixin

class TagsMixin()

[view_source]

Mixin for taggable services

get_by_tags

def get_by_tags(tags: List[str],
                without: bool = False,
                match_all: bool = True) -> List[Dict]

[view_source]

Get all objects with all or any of tags or all objects without any of tags

Class PublishMixin

class PublishMixin()

[view_source]

Mixin for publishable services

publish

def publish(id: int) -> Dict

[view_source]

Publish an object

unpublish

def unpublish(id: int) -> Dict

[view_source]

Unpublish an object

Class NameMixin

class NameMixin()

[view_source]

Mixin for nameable services (by name or slug).

exists

def exists(name_or_slug: str) -> bool

[view_source]

Check if an object with given name_or_slug exists

get_by_name

def get_by_name(name_or_slug: str) -> Optional[Dict]

[view_source]

Get by name or slug

get_id

def get_id(name_or_slug: str) -> Optional[int]

[view_source]

Get the ID for an object by name or slug, returns None if not found

Class PrintMixin

class PrintMixin()

[view_source]

Mixin class for printing data

Class PaginationMixin

class PaginationMixin()

[view_source]

Mixin for paginated services

paginate

def paginate(page: int = 1, page_size: int = 100, **kwargs) -> List[Dict]

[view_source]

Load all items of a paginated resource

fetch_page

def fetch_page(page: int = 1, page_size: int = 100, **kwargs) -> List[Dict]

[view_source]

Load a page of a paginated resource