创建注册任务
(
db: Session,
task_uuid: str,
email_service_id: Optional[int] = None,
proxy: Optional[str] = None
)
| 309 | # ============================================================================ |
| 310 | |
| 311 | def create_registration_task( |
| 312 | db: Session, |
| 313 | task_uuid: str, |
| 314 | email_service_id: Optional[int] = None, |
| 315 | proxy: Optional[str] = None |
| 316 | ) -> RegistrationTask: |
| 317 | """创建注册任务""" |
| 318 | db_task = RegistrationTask( |
| 319 | task_uuid=task_uuid, |
| 320 | email_service_id=email_service_id, |
| 321 | proxy=proxy, |
| 322 | status='pending' |
| 323 | ) |
| 324 | db.add(db_task) |
| 325 | db.commit() |
| 326 | db.refresh(db_task) |
| 327 | return db_task |
| 328 | |
| 329 | |
| 330 | def get_registration_task_by_uuid(db: Session, task_uuid: str) -> Optional[RegistrationTask]: |
nothing calls this directly
no test coverage detected