File size: 3,211 Bytes
89419ea | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
from __future__ import annotations
from datetime import datetime
from typing import Optional, Sequence
from sqlalchemy.orm import Session
from .models import ExternalRegistrationBatch, ExternalRegistrationBatchItem
def create_batch(db: Session, **kwargs) -> ExternalRegistrationBatch:
batch = ExternalRegistrationBatch(**kwargs)
db.add(batch)
db.flush()
db.refresh(batch)
return batch
def create_batch_item(db: Session, **kwargs) -> ExternalRegistrationBatchItem:
item = ExternalRegistrationBatchItem(**kwargs)
db.add(item)
db.flush()
db.refresh(item)
return item
def get_batch_by_uuid(db: Session, batch_uuid: str) -> Optional[ExternalRegistrationBatch]:
return db.query(ExternalRegistrationBatch).filter(ExternalRegistrationBatch.batch_uuid == batch_uuid).first()
def get_batch_by_idempotency_key(db: Session, idempotency_key: str) -> Optional[ExternalRegistrationBatch]:
if not idempotency_key:
return None
return db.query(ExternalRegistrationBatch).filter(ExternalRegistrationBatch.idempotency_key == idempotency_key).first()
def list_batch_items(db: Session, batch_uuid: str) -> list[ExternalRegistrationBatchItem]:
return (
db.query(ExternalRegistrationBatchItem)
.join(ExternalRegistrationBatch, ExternalRegistrationBatch.id == ExternalRegistrationBatchItem.batch_id)
.filter(ExternalRegistrationBatch.batch_uuid == batch_uuid)
.order_by(ExternalRegistrationBatchItem.item_index.asc(), ExternalRegistrationBatchItem.id.asc())
.all()
)
def list_recoverable_batches(db: Session) -> list[ExternalRegistrationBatch]:
return (
db.query(ExternalRegistrationBatch)
.filter(ExternalRegistrationBatch.status.in_(("pending", "running")))
.order_by(ExternalRegistrationBatch.created_at.asc(), ExternalRegistrationBatch.id.asc())
.all()
)
def update_batch(db: Session, batch_uuid: str, **kwargs) -> Optional[ExternalRegistrationBatch]:
batch = get_batch_by_uuid(db, batch_uuid)
if not batch:
return None
for key, value in kwargs.items():
setattr(batch, key, value)
batch.updated_at = datetime.utcnow()
db.flush()
db.refresh(batch)
return batch
def update_batch_item(db: Session, item_id: int, **kwargs) -> Optional[ExternalRegistrationBatchItem]:
item = db.query(ExternalRegistrationBatchItem).filter(ExternalRegistrationBatchItem.id == item_id).first()
if not item:
return None
for key, value in kwargs.items():
setattr(item, key, value)
item.updated_at = datetime.utcnow()
db.flush()
db.refresh(item)
return item
def list_items_by_status(db: Session, batch_uuid: str, statuses: Sequence[str]) -> list[ExternalRegistrationBatchItem]:
return (
db.query(ExternalRegistrationBatchItem)
.join(ExternalRegistrationBatch, ExternalRegistrationBatch.id == ExternalRegistrationBatchItem.batch_id)
.filter(
ExternalRegistrationBatch.batch_uuid == batch_uuid,
ExternalRegistrationBatchItem.status.in_(tuple(statuses)),
)
.order_by(ExternalRegistrationBatchItem.item_index.asc())
.all()
)
|