Add db empty check constraints and migration

Author: piyush-jaiswal

app/models.py

@@ -1,13 +1,20 @@
 from datetime import datetime
 
-from sqlalchemy import Index
+from sqlalchemy import CheckConstraint, Index
 from werkzeug.security import generate_password_hash, check_password_hash
 from email_validator import validate_email, EmailNotValidError
 from email_normalize import normalize
 
 from app import db
 
 
+class ConstraintFactory:
+    @staticmethod
+    def non_empty_string(column_name):
+        constraint_name = f'{column_name}_non_empty'
+        return CheckConstraint(f"TRIM({column_name}) != ''", name=constraint_name)
+
+
 class User(db.Model):
     __tablename__ = 'user'
     id = db.Column(db.Integer, primary_key=True)
@@ -16,6 +23,12 @@ class User(db.Model):
     password_hash = db.Column(db.String(256), nullable=False)
     created_on = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
 
+    __table_args__ = (
+        ConstraintFactory.non_empty_string('email'),
+        ConstraintFactory.non_empty_string('email_normalized'),
+        ConstraintFactory.non_empty_string('password_hash')
+    )
+
     # Does not check for non-deliverable mails. Use check_deliverability or resolve for that which does DNS checks
     # For more stricter validation, use confirmation emails, or a third party API
     @staticmethod
@@ -68,6 +81,10 @@ class Category(db.Model):
     created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
     subcategories = db.relationship("Subcategory", secondary=category_subcategory, back_populates="categories", lazy='dynamic', passive_deletes=True)
 
+    __table_args__ = (
+        ConstraintFactory.non_empty_string('name'),
+    )
+
     def to_json(self):
         return {
             'id': self.id,
@@ -84,6 +101,10 @@ class Subcategory(db.Model):
     categories = db.relationship("Category", secondary=category_subcategory, back_populates="subcategories", lazy='dynamic', passive_deletes=True)
     products = db.relationship("Product", secondary=subcategory_product, back_populates="subcategories", lazy='dynamic', passive_deletes=True)
 
+    __table_args__ = (
+        ConstraintFactory.non_empty_string('name'),
+    )
+
     def to_json(self):
         return {
             'id': self.id,
@@ -100,6 +121,10 @@ class Product(db.Model):
     created_at = db.Column(db.DateTime, nullable=False ,default=datetime.utcnow)
     subcategories = db.relationship("Subcategory", secondary=subcategory_product, back_populates="products", lazy='dynamic', passive_deletes=True)
 
+    __table_args__ = (
+        ConstraintFactory.non_empty_string('name'),
+    )
+
     def to_json(self):
         return {
             'id': self.id,

migrations/versions/911b11318ef1_add_constraints_for_empty_strings.py

@@ -0,0 +1,57 @@
+"""add constraints for empty strings
+
+Revision ID: 911b11318ef1
+Revises: 27b56cc8451c
+Create Date: 2025-09-19 20:25:48.290940
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '911b11318ef1'
+down_revision = '27b56cc8451c'
+branch_labels = None
+depends_on = None
+
+# migrations generated manually! Alembic did not these detect changes.
+
+def upgrade():
+    # User table constraints
+    with op.batch_alter_table('user', schema=None) as batch_op:
+        batch_op.create_check_constraint(batch_op.f('user_email_non_empty_check'), "trim(email) <> ''")
+        batch_op.create_check_constraint(batch_op.f('user_email_normalized_non_empty_check'), "trim(email_normalized) <> ''")
+        batch_op.create_check_constraint(batch_op.f('user_password_hash_non_empty_check'), "trim(password_hash) <> ''")
+
+    # Category table constraint
+    with op.batch_alter_table('category', schema=None) as batch_op:
+        batch_op.create_check_constraint(batch_op.f('category_name_non_empty_check'), "trim(name) <> ''")
+
+    # Subcategory table constraint
+    with op.batch_alter_table('subcategory', schema=None) as batch_op:
+        batch_op.create_check_constraint(batch_op.f('subcategory_name_non_empty_check'), "trim(name) <> ''")
+
+    # Product table constraint
+    with op.batch_alter_table('product', schema=None) as batch_op:
+        batch_op.create_check_constraint(batch_op.f('product_name_non_empty_check'), "trim(name) <> ''")
+
+
+def downgrade():
+    # Product table constraint
+    with op.batch_alter_table('product', schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f('product_name_non_empty_check'), type_='check')
+
+    # Subcategory table constraint
+    with op.batch_alter_table('subcategory', schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f('subcategory_name_non_empty_check'), type_='check')
+
+    # Category table constraint
+    with op.batch_alter_table('category', schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f('category_name_non_empty_check'), type_='check')
+
+    # User table constraints
+    with op.batch_alter_table('user', schema=None) as batch_op:
+        batch_op.drop_constraint(batch_op.f('user_password_hash_non_empty_check'), type_='check')
+        batch_op.drop_constraint(batch_op.f('user_email_normalized_non_empty_check'), type_='check')
+        batch_op.drop_constraint(batch_op.f('user_email_non_empty_check'), type_='check')