Upgrade SQLAlchemy queries to 2.0 standards

[xlorepdarkhelm]

The conversion to SQLAlchemy 2.0 standards continues with a focus on upgrading the old queries to the new way of doing things. Thanks to the DAO principles used in this project, this entire change exists in basically 3 areas:

• The app/dao directory, which contains the query code, isolated from the main project, which is extraordinarily convenient for our needs.
• The tests directory, where some tests are using queries directly and need updating.
• The migrations directory, which has some migrations files using the old queries and will need updating.

Note: SQLAlchemy 2.0 really leans heavily on explicit rather than implicit query structures. As such, you will likely need to explicitly define each join in queries rather than resting on the ORM components providing these for you implicitly.

For an example of a function already converted, go to: app/dao/fact_notification_status_dao.py, the fetch_notification_status_for_service_for_today_and_7_previous_days function.

The process to peform this upgrade is as follows:

Step 1: Refactor SELECT Statements

In SQLAlchemy 2.0, select is used instead of the deprecated query object. This applies to any SELECT operations in the DAOs.

Example: Refactor app/dao Module

Suppose you have a DAO method that retrieves a record based on its ID. The old way using query needs to be updated to the SQLAlchemy 2.0 syntax.

Old Code:

def get_notification_by_id(notification_id):
    return db.session.query(Notification).filter(Notification.id == notification_id).one_or_none()

Updated Code:

from sqlalchemy import select

def get_notification_by_id(notification_id):
    stmt = select(Notification).where(Notification.id == notification_id)
    return db.session.execute(stmt).scalars().one_or_none()

Explanation:

• select: Constructs a query.
• db.session.execute(stmt): Executes the query.
• .scalars(): Extracts the scalar values from the result.
• .one_or_none(): Fetches a single row or returns None.

Step 2: Refactor INSERT Statements

Inserting new records now uses the insert function, where you directly create the insert statement and execute it.

Old Code:

def create_notification(notification):
    db.session.add(notification)
    db.session.commit()

Updated Code:

from sqlalchemy import insert

def create_notification(notification_data):
    stmt = insert(Notification).values(notification_data)
    db.session.execute(stmt)
    db.session.commit()

Explanation:

• insert(Notification): Creates an insert statement for the Notification model.
• .values(notification_data): Populates the insert statement with the provided data.

Step 3: Refactor UPDATE Statements

To update records, use the update statement and replace the old query-based updates.

Old Code:

def update_notification_status(notification_id, status):
    notification = db.session.query(Notification).filter_by(id=notification_id).first()
    notification.status = status
    db.session.commit()

Updated Code:

from sqlalchemy import update

def update_notification_status(notification_id, status):
    stmt = update(Notification).where(Notification.id == notification_id).values(status=status)
    db.session.execute(stmt)
    db.session.commit()

Explanation:

• update(Notification): Specifies that an update will be applied to the Notification model.
• .where(Notification.id == notification_id): Filters the update to only affect the row(s) with the specified notification_id.
• .values(status=status): Specifies the new value for the status column.

Step 4: Refactor DELETE Statements

For deletions, use the delete statement.

Old Code:

def delete_notification(notification_id):
    notification = db.session.query(Notification).filter_by(id=notification_id).first()
    db.session.delete(notification)
    db.session.commit()

Updated Code:

from sqlalchemy import delete

def delete_notification(notification_id):
    stmt = delete(Notification).where(Notification.id == notification_id)
    db.session.execute(stmt)
    db.session.commit()

Explanation:

• delete(Notification): Constructs a delete statement for the Notification model.
• .where(Notification.id == notification_id): Ensures that only the record with the matching ID is deleted.

Step 5: Refactor Utility Functions

If there are utility functions that interact with the database using query, those need to be refactored as well.

Old Utility Code:

def get_active_notifications():
    return db.session.query(Notification).filter(Notification.status == 'active').all()

Updated Utility Code:

from sqlalchemy import select

def get_active_notifications():
    stmt = select(Notification).where(Notification.status == 'active')
    return db.session.execute(stmt).scalars().all()

Step 6: Use Transactions Correctly

SQLAlchemy 2.0 encourages the use of explicit transactions when performing multiple operations. If needed, wrap multiple operations in a transaction block.

Example:

from sqlalchemy.exc import IntegrityError

def create_and_update_notifications(notification_data, notification_id, status):
    try:
        with db.session.begin():  # This starts a transaction block
            # Create a new notification
            stmt_insert = insert(Notification).values(notification_data)
            db.session.execute(stmt_insert)
            
            # Update an existing notification
            stmt_update = update(Notification).where(Notification.id == notification_id).values(status=status)
            db.session.execute(stmt_update)
    except IntegrityError:
        db.session.rollback()
        raise

Step-by-Step Summary:

1. Replace query: Use select statements for querying/getting records.
2. Refactor Inserts: Use insert statements for adding records.
3. Refactor Updates: Use update for modifying records.
4. Refactor Deletes: Use delete statements for removing records.
5. Utility Functions: Any utility code interacting with the database should be upgraded to avoid the old query methods.
6. Transactions: Ensure correct use of transactions using db.session.begin() when performing multiple database operations.