Skip to content

Instantly share code, notes, and snippets.

@dedsm
Created June 11, 2018 12:05
Show Gist options
  • Save dedsm/c724178d43cdcce3efea77aa7bbbffab to your computer and use it in GitHub Desktop.
Save dedsm/c724178d43cdcce3efea77aa7bbbffab to your computer and use it in GitHub Desktop.
Timescale drop_chunks django support
from django.db.models.sql.compiler import SQLCompiler
from django.db.models.sql.constants import NO_RESULTS
class DropChunkCompiler(SQLCompiler):
def as_sql(self):
result = 'SELECT drop_chunks(interval %s, %s)'
return result, [self.query.interval, self.query.db_table]
def execute_sql(self, result_type=NO_RESULTS, *args, **kwargs):
return super().execute_sql(result_type=result_type, *args, **kwargs)
import django
from .compiler import DropChunkCompiler
from .query import DropChunkQuery
class TimescaleManagerMixin(object):
def drop_chunk(self, interval):
query = DropChunkQuery(self.model)
query.set_values(self.model._meta.db_table, interval)
connection = django.db.connections[self.db]
compiler = DropChunkCompiler(query, connection, self.db)
stuff = compiler.execute_sql()
return stuff
from django.db.models import sql
class DropChunkQuery(sql.Query):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.db_table = None
self.interval = None
def set_values(self, db_table, interval):
self.db_table = db_table
self.interval = interval
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment