import os
from chariot import _apis
from chariot.config import getLogger
from chariot.models._utils import _upload_parts
from chariot_api._openapi.video.models import (
InputCreateVideoRequest,
InputCreateUploadRequest,
InputCompleteUploadRequest,
)
__all__ = ["get_videos", "delete_video", "upload_video"]
# defined in go/apps/video/internal/app/upload.go
PART_SIZE = 100 * 1024 * 1024
log = getLogger(__name__)
[docs]
def get_videos(*, project_id: str, to_dict: bool = False, **kwargs):
"""get_videos returns all videos in the requested project_id and kwargs,
optionally converts results to_dict"""
data = _apis.video.videos_api.api_videos_v1_videos_get(project_id=project_id, **kwargs).data
if to_dict:
return [v.to_dict() for v in data]
else:
return data
[docs]
def delete_video(video_id):
"""delete_video deletes a video by video_id"""
return _apis.video.videos_api.api_videos_v1_videos_video_id_delete(video_id)
[docs]
def upload_video(*, project_id: str, name: str, description: str, video_file: str):
"""upload_video uploads the local video located at path `video_file` with the
specified name, description, and project_id"""
log.debug(f"Upload video '{video_file}' to project_id {project_id}")
resp = _apis.video.videos_api.api_videos_v1_videos_post(
video=InputCreateVideoRequest(
project_id=project_id, name=name, description=description, video_type="batch"
)
)
video_id = resp.data.id
log.debug(f"... VideoID: {video_id}")
upload = _apis.video.uploads_api.api_videos_v1_videos_video_id_uploads_post(
video_id=video_id,
upload=InputCreateUploadRequest(
size=os.path.getsize(video_file),
type="video",
),
)
log.debug(
f"... UploadID: {upload.upload_id} with {len(upload.urls)} parts part_size={PART_SIZE}"
)
etags = _upload_parts(video_file, [u.url for u in upload.urls], PART_SIZE)
log.debug(f"... ETags: {etags}")
_apis.video.uploads_api.api_videos_v1_uploads_upload_id_complete_post(
upload_id=upload.upload_id,
request=InputCompleteUploadRequest(
video_id=video_id,
parts=[{"etag": etag, "part_number": idx + 1} for idx, etag in enumerate(etags)],
),
)
return _apis.video.videos_api.api_videos_v1_videos_video_id_get(video_id=video_id).data