sqacodegen strategies in production environments

[doulighan]

Just wondering what strategy people are using to make their production life the most stable.

In a perfect world sqacodegen would output exactly what you need, making the need to manually manage a list of models obsolete. You could stick it in a CI pipeline and feel comfortable that your models are a perfect representation of the database at that point. But of course it doesn't quite work like that, there are going to be modifications you need to make to the output.

Strategy 1) would be to generate your models once, and after that uninstall sqlacodegen from your pyenv relinquish control to the code editor. Any new changes added should be added manually. This feels the most stable, but with dozens of tables and thousands of columns, its really a pain to manage.

2. would be to allow regeneration of the models, but write very particular unit tests ensuring that the proper manual edits have been applied after the fact. Relies on the developers having a good memory. Also feels unstable. But at least huge database changes can be re-imported fairly smoothly.

3. would be to go all-in and create your own generators that give perfect output. I haven't spent any time with the source code, but I'm worried about the time investment, or running into walls that are simply not solvable without massive changes to the source code. In general I have very little experience writing code that outputs other code, I have no idea what I'm in for.

4. Maybe some kind of post-processing, with sed or something? This seems the most complicated to me, especially if you need to add something more dense like methods or hybrid_properties on your models.

Does anyone have opinions on this? How do you set things up?

[amacfie]

I use option 4, but I don't add methods or anything like that, just small changes. I did think about what I'd do if I wanted to add methods, and I thought I'd probably just have functions in a separate module. I assume you've considered that as well, are there advantages to defining methods right on the ORM classes?

[doulighan]

hybrid_properties or validation would be the main benefit. You can then let SQLA perform serialization/deserialization/validation automatically. For big, nested queries trying to manually keep track of columns to apply external functions can be too messy. Although since I posted this, I've found that SQLA's Custom Types are an even less intrusive way to do the same thing.

I actually ended up going with a different version of number 4, using Python's AST module. Its a lot of boilerplate in the beginning, but once you get a feel for it its relatively easy to make broad code changes using the AST, and it feels quite solid when paired with unit testing the output. Adding extra methods to the models would be relatively simple, though Custom Types means I didn't have to.

I'm currently using it to
a) Change datatypes on certain columns
b) Add/remove imports for those datatypes when needed
c) Wrap Table() instances generated from sqlacodegen with Declarative classes, to keep all models using the same interface, regardless of if they're a Table, View, or MaterializedView

[Henkhogan]

@doulighan am make heavy use of extension methods (that also require specific imports) that I want to retain when I update the models. I came up with a (very ugly) python script that merges the model definitions from a staging file (sqlacodegen out) into a destination file from which I import the models in my app. I might me willing to invest a bit more into this approach. Here's the script:

`
source_model_path = 'predictive/models/predictive/staging.py'
extended_model_path = 'predictive/models/predictive/original.py'

MODEL_IMPORTS_HINT = '## model imports'
EXTENSION_HINT = '## extension'

def process_models(model_code: str):
items = []
last_type = None
_class = []
_code_lines = model_code.splitlines()

while _code_lines:
    line = _code_lines.pop(0)

    
    if line.startswith('import') or line.startswith('from'):
        last_type = 'import'
        if _class:
            items.append(('class',_class.copy()))
            _class.clear()
        items.append(('import', line))
        continue

    if line.startswith('del'):
        last_type = 'del'
        if _class:
            items.append(('class',_class.copy()))
            _class.clear()
        items.append(('del', line))
        continue

    if line.startswith('class'):
        last_type = 'class'
        if _class:
            items.append(('class',_class.copy()))
            _class.clear()
        _class.append(line)
        continue

    if last_type == 'class' and (line.startswith(' ') or line.strip()==''):
        _class.append(line)
        continue

    if _class:
        items.append(('class',_class.copy()))
        _class.clear()

    items.append(('any', line))


if _class:
   items.append(('class',_class))

return items

def extract_class_name(line: str):
return line.replace('class ', '').split(':')[0].split('(')[0]

if name == 'main':
with open(source_model_path) as model_file:
source_model_items = process_models(model_file.read())

source_model_imports = [i for t,i in source_model_items if t == 'import']
source_model_classes = {extract_class_name(c[0]): c for t, c in source_model_items if t == 'class'}


with open(extended_model_path) as model_file:
    content = model_file.read()

extended_model_items = process_models(content)
working_model_items = process_models(content)

dotted_model_path = source_model_path.replace('.py','').replace('/','.')

extended_model_classes = {extract_class_name(c[0]): c for t, c in extended_model_items if t == 'class'}

extension_class = {}
extension_attributes = {}
for _classname, _class in extended_model_classes.items():
    _on = False
    for line in _class:
        if EXTENSION_HINT in line:
            _on = True
            extension_class[_classname] = _class[0]
        if _on:
            extension_attributes.setdefault(_classname, []).append(line)


##clear source model imports
_model_imports = False
_model_import_start_index = None
_model_import_end_index = None
_index = -1
for t, i in working_model_items:
    _index += 1
    if i.strip() == '' and _model_imports:
        _model_import_end_index = _index -1
        break

    if i.strip() == MODEL_IMPORTS_HINT:
        _model_imports = True
        _model_import_start_index = _index
        continue

for _ in range(_model_import_end_index, _model_import_start_index, -1):
  working_model_items.pop(_) 
        

for i in reversed(source_model_imports):
    working_model_items.insert(_model_import_start_index+1, ('import', i))


for t, c in working_model_items:
    if t == 'class':
        class_name = extract_class_name(c[0])
        if class_name in source_model_classes:
            c.clear()
            c += source_model_classes[class_name]
            if class_name in extension_class:
                c[0] = extension_class[class_name]
                c += extension_attributes[class_name]

with open(extended_model_path, 'w') as dest:
    for t,i in working_model_items:
        if t == 'class':
            for line in i:
                dest.write(f'{line}\n')
        else:
            dest.write(f'{i}\n')

`