53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
from sqlalchemy.orm import Session
|
|
from . import models, schemas
|
|
|
|
def get_user(db: Session, user_id: int):
|
|
return db.query(models.User).filter(models.User.id == user_id).first()
|
|
|
|
def get_user_by_username(db: Session, username: str):
|
|
return db.query(models.User).filter(models.User.username == username).first()
|
|
|
|
def get_users(db: Session, skip: int = 0, limit: int = 100):
|
|
return db.query(models.User).offset(skip).limit(limit).all()
|
|
|
|
def create_user(db: Session, user: schemas.UserCreate, hashed_password: str):
|
|
db_user = models.User(username=user.username, hashed_password=hashed_password, role=user.role)
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
return db_user
|
|
|
|
def delete_user(db: Session, user_id: int):
|
|
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
|
if db_user:
|
|
db.delete(db_user)
|
|
db.commit()
|
|
return db_user
|
|
|
|
def get_jobs(db: Session, skip: int = 0, limit: int = 100, user_id: int = None):
|
|
if user_id:
|
|
return db.query(models.Job).filter(models.Job.user_id == user_id).order_by(models.Job.created_at.desc()).offset(skip).limit(limit).all()
|
|
return db.query(models.Job).order_by(models.Job.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
def get_job(db: Session, job_id: int):
|
|
return db.query(models.Job).filter(models.Job.id == job_id).first()
|
|
|
|
def create_job(db: Session, job: schemas.JobCreate, user_id: int):
|
|
db_job = models.Job(**job.dict(), user_id=user_id)
|
|
db.add(db_job)
|
|
db.commit()
|
|
db.refresh(db_job)
|
|
return db_job
|
|
|
|
def update_job_status(db: Session, job_id: int, status: str, result_path: str = None, duration: str = None):
|
|
job = db.query(models.Job).filter(models.Job.id == job_id).first()
|
|
if job:
|
|
job.status = status
|
|
if result_path:
|
|
job.result_path = result_path
|
|
if duration:
|
|
job.duration = duration
|
|
db.commit()
|
|
db.refresh(job)
|
|
return job
|