class MyPostgresPersistence(IdentifiablePostgresPersistence):
def __init__(self):
super(MyPostgresPersistence, self).__init__('mydata')
def _define_schema(self):
# clear all previously autogenerated schemas
self._clear_schema()
# create a table
self._ensure_schema('CREATE TABLE ' + self._table_name + ' (id TEXT PRIMARY KEY, key TEXT, content TEXT)')
# create an index
self._ensure_index(self._table_name + '_key', {'key': 1}, {'unique': True})
def get_one_random(self, correlation_id: Optional[str], filter: Any) -> MyData:
return super().get_one_random(correlation_id, filter)
def get_list_by_filter(self, correlation_id: Optional[str], filter: Any, sort: Any, select: Any) -> List[MyData]:
return super().get_list_by_filter(correlation_id, filter, sort, select)
def get_count_by_filter(self, correlation_id: Optional[str], filter: Any) -> int:
return super().get_count_by_filter(correlation_id, filter)
def get_page_by_filter(self, correlation_id: Optional[str], filter: Any, paging: PagingParams, sort: Any, select: Any) -> DataPage:
return super().get_page_by_filter(correlation_id, filter, paging, sort, select)
def delete_by_filter(self, correlation_id: Optional[str], filter: Any) -> int:
return super().delete_by_filter(correlation_id, filter)