68 lines
1.5 KiB
Python
68 lines
1.5 KiB
Python
from typing import Protocol
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .models import OtpMessageLog, SmsMessageLog, UserAccount
|
|
|
|
|
|
class SendResultLike(Protocol):
|
|
to: str
|
|
success: bool
|
|
job_id: str | None
|
|
message_id: str | None
|
|
error: str | None
|
|
|
|
|
|
def record_sms_logs(
|
|
db: Session,
|
|
user: UserAccount,
|
|
*,
|
|
sender_id: str,
|
|
template_name: str,
|
|
template_id: UUID,
|
|
results: list[SendResultLike],
|
|
) -> None:
|
|
for item in results:
|
|
db.add(
|
|
SmsMessageLog(
|
|
user_id=user.id,
|
|
sender_id=sender_id,
|
|
to_number=item.to,
|
|
template_name=template_name,
|
|
template_id=template_id,
|
|
status="sent" if item.success else "failed",
|
|
job_id=item.job_id,
|
|
message_id=item.message_id,
|
|
error_message=item.error,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def record_otp_log(
|
|
db: Session,
|
|
user: UserAccount,
|
|
*,
|
|
project_name: str,
|
|
template_type: str,
|
|
to_number: str,
|
|
status: str,
|
|
job_id: str | None = None,
|
|
message_id: str | None = None,
|
|
error_message: str | None = None,
|
|
) -> OtpMessageLog:
|
|
entry = OtpMessageLog(
|
|
user_id=user.id,
|
|
project_name=project_name,
|
|
template_type=template_type,
|
|
to_number=to_number,
|
|
status=status,
|
|
job_id=job_id,
|
|
message_id=message_id,
|
|
error_message=error_message,
|
|
)
|
|
db.add(entry)
|
|
db.commit()
|
|
db.refresh(entry)
|
|
return entry
|