CODEWITHSUNDEEP
pythonpostgresqlrestflaskflask-sqlalchemy
Karen
Karen

Reputation: 29

Using flask-sqlalchemy, how to specify user-defined data type in database table column

I am writing a rest api service in python that talks to a Postgres database. I am using flake-sqlalchemy to do so. In the table, one of the columns is defined as:

my_column my_type not null

And my_type is defined:

create type my_type as emum ('val1', 'val2', 'val3')

How do I represent this in flask-sqlalchemy? Thanks you for you help.

Upvotes: 0

Views: 596

Answers (1)

Peter Moore
Peter Moore

Reputation: 2126

your app.py definition goes like this:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy.types import TypeDecorator, CHAR

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://test:test@localhost/test?charset=utf8mb4' 
db = SQLAlchemy(app)
Migrate(app, db, compare_type=True)

class CustEnum(TypeDecorator):
    impl = CHAR(4)
    def process_bind_param(self, value, dialect):
        valid_values = ('val1', 'val2', 'val3')
        if value not in valid_values:
            raise TypeError(f"mismatch {value} not in {valid_values} ")
            
        return value

class Blobby(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    type_col = db.Column(CustEnum())

where the process_bind_param method is basically the setter method and process_result_value is the getter method.

refer to doc for more on it https://docs.sqlalchemy.org/en/14/core/custom_types.html#sqlalchemy.types.TypeDecorator

I also like to add a test function to the file with the cli decorator to check it works.

@app.cli.command("testit")
def testit():
    newblob = Blobby(type_col = 'val2')
    db.session.add( newblob )
    db.session.commit()
    
    for row in db.session.execute('select * from blobby'):
        print(row)
    
    input('press enter to continue to failure test')
    # value not allowed should fail
    newblob = Blobby(type_col = 'fail')
    db.session.add( newblob )
    db.session.commit()

When you apply the migration (i.e running flask db init, flask db migrate and flask db upgrade), you get an error that name app.CustEnum cant be found.

sa.Column('type_col', app.CustEnum(length=4), nullable=True),
NameError: name 'app' is not defined

Searching for a fix for that I found the author of the library says you need to update the migration script itself which is kinda backwards IMO.

https://github.com/miguelgrinberg/Flask-Migrate/issues/282#issuecomment-518989508

so just edit migrations/versions/98f29efa609f_.py or whatever your version is. and add import app or whatever your flask script's namespace is.

last thing is to run the test with flask testit to run the testit() function to give you one success and one failure.

sqlalchemy.exc.StatementError: (builtins.TypeError) mismatch fail not in ('val1', 'val2', 'val3')
[SQL: INSERT INTO blobby (type_col) VALUES (%(type_col)s)]
[parameters: [{'type_col': 'fail'}]]

Upvotes: 1

Related Questions