crud_video.py 1.3 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344
  1. from typing import List
  2. from fastapi.encoders import jsonable_encoder
  3. from sqlalchemy.orm import Session
  4. from sqlalchemy import desc
  5. from app.crud.base import CRUDBase
  6. from app.models.video import Video
  7. from app.schemas.video import VideoCreate, VideoUpdate
  8. from app.utils import random_name
  9. class CRUDVideo(CRUDBase[Video, VideoCreate, VideoUpdate]):
  10. def create_with_owner(
  11. self, db: Session, *, obj_in: VideoCreate, owner_id: int
  12. ) -> Video:
  13. obj_in_data = jsonable_encoder(obj_in)
  14. db_obj = self.model(**obj_in_data, owner_id=owner_id)
  15. db.add(db_obj)
  16. db.commit()
  17. db.refresh(db_obj)
  18. return db_obj
  19. def get_multi_by_owner(
  20. self, db: Session, *, owner_id: int, skip: int = 0, limit: int = 100
  21. ) -> List[Video]:
  22. return (
  23. db.query(self.model)
  24. .filter(Video.owner_id == owner_id)
  25. .order_by(desc(Video.id))
  26. .offset(skip)
  27. .limit(limit)
  28. .all()
  29. )
  30. def generate_file_name(
  31. self, db: Session, *, n:int
  32. ) -> bool:
  33. while True:
  34. filename = random_name(n)
  35. if not db.query(self.model).filter(Video.stored_filename==filename).first():
  36. return filename
  37. video = CRUDVideo(Video)