"""Generate the errors module from PostgreSQL source code.
The script can be run at a new PostgreSQL release to refresh the module.
"""
import os
import re
import sys
from urllib.request import urlopen
from collections import defaultdict
def main():
filename = os.path.join(
os.path.dirname(__file__), "../psycopg/sqlstate_errors.h")
classes, errors = fetch_errors(
['9.1', '9.2', '9.3', '9.4', '9.5', '9.6', '10', '11', '12', '13'])
f = open(filename, "w")
print("/*\n * Autogenerated by 'scripts/make_errors.py'.\n */\n", file=f)
for line in generate_module_data(classes, errors):
print(line, file=f)
def parse_errors_txt(url):
classes = {}
errors = defaultdict(dict)
page = urlopen(url)
for line in page:
line = line.decode('ascii').split('#')[0].strip()
if not line:
continue
m = re.match(r"Section: (Class (..) - .+)", line)
if m:
label, class_ = m.groups()
classes[class_] = label
continue
m = re.match(r"(.....)\s+(?:E|W|S)\s+ERRCODE_(\S+)(?:\s+(\S+))?$", line)
if m:
errcode, macro, spec = m.groups()
if not spec:
continue
errlabel = spec.upper()
errors[class_][errcode] = errlabel
continue
raise ValueError(f"unexpected line:\n{line}")
return classes, errors
errors_txt_url = \
"http://git.postgresql.org/gitweb/?p=postgresql.git;a=blob_plain;" \
"f=src/backend/utils/errcodes.txt;hb=%s"
def fetch_errors(versions):
classes = {}
errors = defaultdict(dict)
for version in versions:
print(version, file=sys.stderr)
tver = tuple(map(int, version.split()[0].split('.')))
tag = f"{tver[0] >= 10 and 'REL_' or 'REL'}{version.replace('.', '_')}_STABLE"
c1, e1 = parse_errors_txt(errors_txt_url % tag)
classes.update(c1)
for c, cerrs in e1.items():
errors[c].update(cerrs)
return classes, errors
def generate_module_data(classes, errors):
tmpl = '{"%(errcode)s", "%(cls)s"},'
specific = {
'38002': 'ModifyingSqlDataNotPermittedExt',
'38003': 'ProhibitedSqlStatementAttemptedExt',
'38004': 'ReadingSqlDataNotPermittedExt',
'39004': 'NullValueNotAllowedExt',
'XX000': 'InternalError_',
}
seen = set("""
Error Warning InterfaceError DataError DatabaseError ProgrammingError
IntegrityError InternalError NotSupportedError OperationalError
QueryCanceledError TransactionRollbackError
""".split())
for clscode, clslabel in sorted(classes.items()):
if clscode in ('00', '01'):
continue
yield f"\n/* {clslabel} */"
for errcode, errlabel in sorted(errors[clscode].items()):
if errcode in specific:
clsname = specific[errcode]
else:
clsname = errlabel.title().replace('_', '')
if clsname in seen:
raise Exception(f"class already existing: {clsname}")
seen.add(clsname)
yield tmpl % {
'cls': clsname,
'errcode': errcode
}
if __name__ == '__main__':
sys.exit(main())