darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-02 09:26:08 -05:00
Code Issues Packages Projects Releases Wiki Activity

start InboundSmsDao

Browse Source
This commit is contained in:
Kenneth Kehl
2024-10-17 12:24:23 -07:00
parent bc9ff7405f
commit c818bac288
1 changed files with 10 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
app/dao/inbound_sms_dao.py
View File

@@ -1,7 +1,7 @@
from flask import current_app from flask import current_app
from sqlalchemy import and_, desc from sqlalchemy import and_, desc
from sqlalchemy.dialects.postgresql import insert from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import aliased from sqlalchemy.orm import aliased, select
from app import db from app import db
from app.dao.dao_utils import autocommit from app.dao.dao_utils import autocommit
@@ -18,8 +18,10 @@ def dao_create_inbound_sms(inbound_sms):
def dao_get_inbound_sms_for_service( def dao_get_inbound_sms_for_service(
service_id, user_number=None, *, limit_days=None, limit=None service_id, user_number=None, *, limit_days=None, limit=None
): ):
q = InboundSms.query.filter(InboundSms.service_id == service_id).order_by( q = (
InboundSms.created_at.desc() select(InboundSms)
.filter(InboundSms.service_id == service_id)
.order_by(InboundSms.created_at.desc())
) )
if limit_days is not None: if limit_days is not None:
start_date = midnight_n_days_ago(limit_days) start_date = midnight_n_days_ago(limit_days)
@@ -31,7 +33,7 @@ def dao_get_inbound_sms_for_service(
if limit: if limit:
q = q.limit(limit) q = q.limit(limit)
return q.all() return db.session.execute(q).scalars().all()
def dao_get_paginated_inbound_sms_for_service_for_public_api( def dao_get_paginated_inbound_sms_for_service_for_public_api(
@@ -58,10 +60,12 @@ def dao_get_paginated_inbound_sms_for_service_for_public_api(
def dao_count_inbound_sms_for_service(service_id, limit_days): def dao_count_inbound_sms_for_service(service_id, limit_days):
return InboundSms.query.filter( stmt = select(InboundSms).filter(
InboundSms.service_id == service_id, InboundSms.service_id == service_id,
InboundSms.created_at >= midnight_n_days_ago(limit_days), InboundSms.created_at >= midnight_n_days_ago(limit_days),
).count() )
result = db.session.execute(stmt)
return result.rowcount
def _insert_inbound_sms_history(subquery, query_limit=10000): def _insert_inbound_sms_history(subquery, query_limit=10000):