fix more sqlalchemy

.ds.baseline

@@ -305,7 +305,7 @@
         "filename": "tests/app/service/test_rest.py",
         "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8",
         "is_verified": false,
-        "line_number": 1284,
+        "line_number": 1285,
         "is_secret": false
       }
     ],
@@ -341,15 +341,15 @@
         "filename": "tests/app/user/test_rest.py",
         "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8",
         "is_verified": false,
-        "line_number": 108,
+        "line_number": 110,
         "is_secret": false
       },
       {
         "type": "Secret Keyword",
         "filename": "tests/app/user/test_rest.py",
         "hashed_secret": "0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33",
         "is_verified": false,
-        "line_number": 826,
+        "line_number": 858,
         "is_secret": false
       }
     ],
@@ -384,5 +384,5 @@
       }
     ]
   },
-  "generated_at": "2024-10-31T21:25:32Z"
+  "generated_at": "2024-11-15T18:43:06Z"
 }

app/billing/rest.py

@@ -1,5 +1,6 @@
 from flask import Blueprint, jsonify, request
 
+from app import db
 from app.billing.billing_schemas import (
     create_or_update_free_sms_fragment_limit_schema,
     serialize_ft_billing_remove_emails,
@@ -60,7 +61,7 @@ def get_free_sms_fragment_limit(service_id):
     )
 
     if annual_billing is None:
-        service = Service.query.get(service_id)
+        service = db.session.get(Service, service_id)
         # An entry does not exist in annual_billing table for that service and year.
         # Set the annual billing to the default free allowance based on the organization type of the service.
 

app/celery/scheduled_tasks.py

@@ -1,10 +1,10 @@
 from datetime import timedelta
 
 from flask import current_app
-from sqlalchemy import between
+from sqlalchemy import between, select, union, update
 from sqlalchemy.exc import SQLAlchemyError
 
-from app import notify_celery, zendesk_client
+from app import db, notify_celery, zendesk_client
 from app.celery.tasks import (
     get_recipient_csv_and_template_and_sender_id,
     process_incomplete_jobs,
@@ -18,7 +18,6 @@
 from app.dao.invited_user_dao import expire_invitations_created_more_than_two_days_ago
 from app.dao.jobs_dao import (
     dao_set_scheduled_jobs_to_pending,
-    dao_update_job,
     find_jobs_with_missing_rows,
     find_missing_row_for_job,
 )
@@ -105,30 +104,32 @@ def check_job_status():
     thirty_minutes_ago = utc_now() - timedelta(minutes=30)
     thirty_five_minutes_ago = utc_now() - timedelta(minutes=35)
 
-    incomplete_in_progress_jobs = Job.query.filter(
+    incomplete_in_progress_jobs = select(Job).filter(
         Job.job_status == JobStatus.IN_PROGRESS,
         between(Job.processing_started, thirty_five_minutes_ago, thirty_minutes_ago),
     )
-    incomplete_pending_jobs = Job.query.filter(
+    incomplete_pending_jobs = select(Job).filter(
         Job.job_status == JobStatus.PENDING,
         Job.scheduled_for.isnot(None),
         between(Job.scheduled_for, thirty_five_minutes_ago, thirty_minutes_ago),
     )
 
-    jobs_not_complete_after_30_minutes = (
-        incomplete_in_progress_jobs.union(incomplete_pending_jobs)
-        .order_by(Job.processing_started, Job.scheduled_for)
-        .all()
-    )
+    jobs_not_complete_after_30_minutes = db.session.execute(
+        union(incomplete_in_progress_jobs, incomplete_pending_jobs).order_by(
+            Job.processing_started, Job.scheduled_for
+        )
+    ).all()
 
     # temporarily mark them as ERROR so that they don't get picked up by future check_job_status tasks
     # if they haven't been re-processed in time.
     job_ids = []
     for job in jobs_not_complete_after_30_minutes:
-        job.job_status = JobStatus.ERROR
-        dao_update_job(job)
-        job_ids.append(str(job.id))
+        db.session.execute(
+            update(Job).where(Job.id == job.id).values(job_status=JobStatus.ERROR)
+        )
+        db.session.commit()
 
+        job_ids.append(str(job.id))
     if job_ids:
         current_app.logger.info("Job(s) {} have not completed.".format(job_ids))
         process_incomplete_jobs.apply_async([job_ids], queue=QueueNames.JOBS)

app/dao/api_key_dao.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import timedelta
 
-from sqlalchemy import func, or_
+from sqlalchemy import func, or_, select
 
 from app import db
 from app.dao.dao_utils import autocommit, version_class
@@ -23,31 +23,61 @@ def save_model_api_key(api_key):
 @autocommit
 @version_class(ApiKey)
 def expire_api_key(service_id, api_key_id):
-    api_key = ApiKey.query.filter_by(id=api_key_id, service_id=service_id).one()
+    api_key = (
+        db.session.execute(
+            select(ApiKey).where(
+                ApiKey.id == api_key_id, ApiKey.service_id == service_id
+            )
+        )
+        .scalars()
+        .one()
+    )
     api_key.expiry_date = utc_now()
     db.session.add(api_key)
 
 
 def get_model_api_keys(service_id, id=None):
     if id:
-        return ApiKey.query.filter_by(
-            id=id, service_id=service_id, expiry_date=None
-        ).one()
+        return (
+            db.session.execute(
+                select(ApiKey).where(
+                    ApiKey.id == id,
+                    ApiKey.service_id == service_id,
+                    ApiKey.expiry_date == None, # noqa
+                )
+            )
+            .scalars()
+            .one()
+        )
     seven_days_ago = utc_now() - timedelta(days=7)
-    return ApiKey.query.filter(
-        or_(
-            ApiKey.expiry_date == None,  # noqa
-            func.date(ApiKey.expiry_date) > seven_days_ago,  # noqa
-        ),
-        ApiKey.service_id == service_id,
-    ).all()
+    return (
+        db.session.execute(
+            select(ApiKey).where(
+                or_(
+                    ApiKey.expiry_date == None,  # noqa
+                    func.date(ApiKey.expiry_date) > seven_days_ago,  # noqa
+                ),
+                ApiKey.service_id == service_id,
+            )
+        )
+        .scalars()
+        .all()
+    )
 
 
 def get_unsigned_secrets(service_id):
     """
     This method can only be exposed to the Authentication of the api calls.
     """
-    api_keys = ApiKey.query.filter_by(service_id=service_id, expiry_date=None).all()
+    api_keys = (
+        db.session.execute(
+            select(ApiKey).where(
+                ApiKey.service_id == service_id, ApiKey.expiry_date == None  # noqa
+            )
+        )
+        .scalars()
+        .all()
+    )
     keys = [x.secret for x in api_keys]
     return keys
 
@@ -56,5 +86,11 @@ def get_unsigned_secret(key_id):
     """
     This method can only be exposed to the Authentication of the api calls.
     """
-    api_key = ApiKey.query.filter_by(id=key_id, expiry_date=None).one()
+    api_key = (
+        db.session.execute(
+            select(ApiKey).where(ApiKey.id == key_id, ApiKey.expiry_date == None)  # noqa
+        )
+        .scalars()
+        .one()
+    )
     return api_key.secret

app/dao/email_branding_dao.py

@@ -1,18 +1,28 @@
+from sqlalchemy import select
+
 from app import db
 from app.dao.dao_utils import autocommit
 from app.models import EmailBranding
 
 
 def dao_get_email_branding_options():
-    return EmailBranding.query.all()
+    return db.session.execute(select(EmailBranding)).scalars().all()
 
 
 def dao_get_email_branding_by_id(email_branding_id):
-    return EmailBranding.query.filter_by(id=email_branding_id).one()
+    return (
+        db.session.execute(select(EmailBranding).filter_by(id=email_branding_id))
+        .scalars()
+        .one()
+    )
 
 
 def dao_get_email_branding_by_name(email_branding_name):
-    return EmailBranding.query.filter_by(name=email_branding_name).first()
+    return (
+        db.session.execute(select(EmailBranding).filter_by(name=email_branding_name))
+        .scalars()
+        .first()
+    )
 
 
 @autocommit