H9vn5uU1

1import re
2
3from celery import shared_task
4from django.db import connection
5from psycopg2.extensions import AsIs
6
7CHECK_VALUE_RE = re.compile(r'^[A-Z,a-z,0-9,_,.]*$')
8
9
10def check_value(value):
11    if CHECK_VALUE_RE.search(value) is None:
12        raise ValueError("Incorrect value for table of field name: %s" % value)
13
14
15@shared_task
16def create_pg_index(table_name, field):
17    check_value(table_name)
18    check_value(field)
19    fields = field.split('.')
20    index_name = "%s_%s_idx" % (table_name, "_".join(fields))
21    selector = " -> '%s'" * len(fields) % tuple(fields)
22    with connection.cursor() as cursor:
23        params = {
24            'index_name': AsIs(index_name),
25            'table_name': AsIs(table_name),
26            'selector': AsIs(selector),
27        }
28        cursor.execute(
29            """CREATE INDEX IF NOT EXISTS %(index_name)s ON %(table_name)s USING gin (( data %(selector)s));""",
30            params
31        )