set_trigger.py
1.39 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import sys
from configparser import ConfigParser
from argparse import ArgumentParser
from sqlalchemy import (
engine_from_config,
exc,
)
from sqlalchemy.sql.expression import text
def get_option(argv):
default_schema = 'public'
help_schema = 'default ' + default_schema
pars = ArgumentParser()
pars.add_argument('conf')
pars.add_argument('--table', required=True)
pars.add_argument('--schema', default=default_schema, help=help_schema)
return pars.parse_args(argv)
def main(argv=sys.argv[1:]):
option = get_option(argv)
conf_file = option.conf
conf = ConfigParser()
conf.read(conf_file)
engine = engine_from_config(conf['main'], 'db_')
engine.echo = True
table_name = option.table
table_schema = option.schema
if table_schema:
table_name = '.'.join([table_schema, table_name])
for operation in ('insert', 'update', 'delete'):
trigger_name = f'log_table_{operation}'
sql = f'DROP TRIGGER {trigger_name} ON {table_name}'
try:
engine.execute(text(sql))
except exc.ProgrammingError as e:
s = str(e)
if s.find('does not exist') == -1:
raise e
sql = f'CREATE TRIGGER {trigger_name} '\
f'AFTER {operation} ON {table_name} '\
f'FOR EACH ROW EXECUTE PROCEDURE log_table_after_{operation}()'
engine.execute(text(sql))