Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions videodb/_constants.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""Constants used in the videodb package."""

from typing import Union
from dataclasses import dataclass

Expand Down
24 changes: 23 additions & 1 deletion videodb/audio.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,17 @@


class Audio:
def __init__(self, _connection, id: str, collection_id: str, **kwargs) -> None:
"""Audio class to interact with the Audio

:ivar str id: Unique identifier for the audio
:ivar str collection_id: ID of the collection this audio belongs to
:ivar str name: Name of the audio file
:ivar float length: Duration of the audio in seconds
"""

def __init__(
self, _connection, id: str, collection_id: str, **kwargs
) -> None:
self._connection = _connection
self.id = id
self.collection_id = collection_id
Expand All @@ -21,11 +31,23 @@ def __repr__(self) -> str:
)

def generate_url(self) -> str:
"""Generate the signed url of the audio.

:raises InvalidRequestError: If the get_url fails
:return: The signed url of the audio
:rtype: str
"""
url_data = self._connection.post(
path=f"{ApiPath.audio}/{self.id}/{ApiPath.generate_url}",
params={"collection_id": self.collection_id},
)
return url_data.get("signed_url", None)

def delete(self) -> None:
"""Delete the audio.

:raises InvalidRequestError: If the delete fails
:return: None if the delete is successful
:rtype: None
"""
self._connection.delete(f"{ApiPath.audio}/{self.id}")
70 changes: 69 additions & 1 deletion videodb/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,32 @@


class Connection(HttpClient):
def __init__(self, api_key: str, base_url: str) -> None:
"""Connection class to interact with the VideoDB"""

def __init__(self, api_key: str, base_url: str) -> "Connection":
"""Initializes a new instance of the Connection class with specified API credentials.

Note: Users should not initialize this class directly.
Instead use :meth:`videodb.connect() <videodb.connect>`

:param str api_key: API key for authentication
:param str base_url: Base URL of the VideoDB API
:raise ValueError: If the API key is not provided
:return: :class:`Connection <Connection>` object, to interact with the VideoDB
:rtype: :class:`videodb.client.Connection`
"""
self.api_key = api_key
self.base_url = base_url
self.collection_id = "default"
super().__init__(api_key=api_key, base_url=base_url, version=__version__)

def get_collection(self, collection_id: Optional[str] = "default") -> Collection:
"""Get a collection object by its ID.

:param str collection_id: ID of the collection (optional, default: "default")
:return: :class:`Collection <Collection>` object
:rtype: :class:`videodb.collection.Collection`
"""
collection_data = self.get(path=f"{ApiPath.collection}/{collection_id}")
self.collection_id = collection_data.get("id", "default")
return Collection(
Expand All @@ -42,6 +61,11 @@ def get_collection(self, collection_id: Optional[str] = "default") -> Collection
)

def get_collections(self) -> List[Collection]:
"""Get a list of all collections.

:return: List of :class:`Collection <Collection>` objects
:rtype: list[:class:`videodb.collection.Collection`]
"""
collections_data = self.get(path=ApiPath.collection)
return [
Collection(
Expand All @@ -57,6 +81,14 @@ def get_collections(self) -> List[Collection]:
def create_collection(
self, name: str, description: str, is_public: bool = False
) -> Collection:
"""Create a new collection.

:param str name: Name of the collection
:param str description: Description of the collection
:param bool is_public: Make collection public (optional, default: False)
:return: :class:`Collection <Collection>` object
:rtype: :class:`videodb.collection.Collection`
"""
collection_data = self.post(
path=ApiPath.collection,
data={
Expand All @@ -75,6 +107,14 @@ def create_collection(
)

def update_collection(self, id: str, name: str, description: str) -> Collection:
"""Update an existing collection.

:param str id: ID of the collection
:param str name: Name of the collection
:param str description: Description of the collection
:return: :class:`Collection <Collection>` object
:rtype: :class:`videodb.collection.Collection`
"""
collection_data = self.patch(
path=f"{ApiPath.collection}/{id}",
data={
Expand All @@ -92,12 +132,29 @@ def update_collection(self, id: str, name: str, description: str) -> Collection:
)

def check_usage(self) -> dict:
"""Check the usage.

:return: Usage data
:rtype: dict
"""
return self.get(path=f"{ApiPath.billing}/{ApiPath.usage}")

def get_invoices(self) -> List[dict]:
"""Get a list of all invoices.

:return: List of invoices
:rtype: list[dict]
"""
return self.get(path=f"{ApiPath.billing}/{ApiPath.invoices}")

def download(self, stream_link: str, name: str) -> dict:
"""Download a file from a stream link.

:param stream_link: URL of the stream to download
:param name: Name to save the downloaded file as
:return: Download response data
:rtype: dict
"""
return self.post(
path=f"{ApiPath.download}",
data={
Expand All @@ -115,6 +172,17 @@ def upload(
description: Optional[str] = None,
callback_url: Optional[str] = None,
) -> Union[Video, Audio, Image, None]:
"""Upload a file.

:param str file_path: Path to the file to upload (optional)
:param str url: URL of the file to upload (optional)
:param MediaType media_type: MediaType object (optional)
:param str name: Name of the file (optional)
:param str description: Description of the file (optional)
:param str callback_url: URL to receive the callback (optional)
:return: :class:`Video <Video>`, or :class:`Audio <Audio>`, or :class:`Image <Image>` object
:rtype: Union[ :class:`videodb.video.Video`, :class:`videodb.audio.Audio`, :class:`videodb.image.Image`]
"""
upload_data = upload(
self,
file_path,
Expand Down
88 changes: 87 additions & 1 deletion videodb/collection.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,12 @@


class Collection:
"""Collection class to interact with the Collection.

Note: Users should not initialize this class directly.
Instead use :meth:`Connection.get_collection() <videodb.client.Connection.get_collection>`
"""

def __init__(
self,
_connection,
Expand Down Expand Up @@ -57,20 +63,31 @@ def delete(self) -> None:
self._connection.delete(path=f"{ApiPath.collection}/{self.id}")

def get_videos(self) -> List[Video]:
"""Get all the videos in the collection.

:return: List of :class:`Video <Video>` objects
:rtype: List[:class:`videodb.video.Video`]
"""
videos_data = self._connection.get(
path=f"{ApiPath.video}",
params={"collection_id": self.id},
)
return [Video(self._connection, **video) for video in videos_data.get("videos")]

def get_video(self, video_id: str) -> Video:
"""Get a video by its ID.

:param str video_id: ID of the video
:return: :class:`Video <Video>` object
:rtype: :class:`videodb.video.Video`
"""
video_data = self._connection.get(
path=f"{ApiPath.video}/{video_id}", params={"collection_id": self.id}
)
return Video(self._connection, **video_data)

def delete_video(self, video_id: str) -> None:
"""Delete the video
"""Delete the video.

:param str video_id: The id of the video to be deleted
:raises InvalidRequestError: If the delete fails
Expand All @@ -82,37 +99,73 @@ def delete_video(self, video_id: str) -> None:
)

def get_audios(self) -> List[Audio]:
"""Get all the audios in the collection.

:return: List of :class:`Audio <Audio>` objects
:rtype: List[:class:`videodb.audio.Audio`]
"""
audios_data = self._connection.get(
path=f"{ApiPath.audio}",
params={"collection_id": self.id},
)
return [Audio(self._connection, **audio) for audio in audios_data.get("audios")]

def get_audio(self, audio_id: str) -> Audio:
"""Get an audio by its ID.

:param str audio_id: ID of the audio
:return: :class:`Audio <Audio>` object
:rtype: :class:`videodb.audio.Audio`
"""
audio_data = self._connection.get(
path=f"{ApiPath.audio}/{audio_id}", params={"collection_id": self.id}
)
return Audio(self._connection, **audio_data)

def delete_audio(self, audio_id: str) -> None:
"""Delete the audio.

:param str audio_id: The id of the audio to be deleted
:raises InvalidRequestError: If the delete fails
:return: None if the delete is successful
:rtype: None
"""
return self._connection.delete(
path=f"{ApiPath.audio}/{audio_id}", params={"collection_id": self.id}
)

def get_images(self) -> List[Image]:
"""Get all the images in the collection.

:return: List of :class:`Image <Image>` objects
:rtype: List[:class:`videodb.image.Image`]
"""
images_data = self._connection.get(
path=f"{ApiPath.image}",
params={"collection_id": self.id},
)
return [Image(self._connection, **image) for image in images_data.get("images")]

def get_image(self, image_id: str) -> Image:
"""Get an image by its ID.

:param str image_id: ID of the image
:return: :class:`Image <Image>` object
:rtype: :class:`videodb.image.Image`
"""
image_data = self._connection.get(
path=f"{ApiPath.image}/{image_id}", params={"collection_id": self.id}
)
return Image(self._connection, **image_data)

def delete_image(self, image_id: str) -> None:
"""Delete the image.

:param str image_id: The id of the image to be deleted
:raises InvalidRequestError: If the delete fails
:return: None if the delete is successful
:rtype: None
"""
return self._connection.delete(
path=f"{ApiPath.image}/{image_id}", params={"collection_id": self.id}
)
Expand All @@ -127,6 +180,18 @@ def search(
dynamic_score_percentage: Optional[float] = None,
filter: List[Dict[str, Any]] = [],
) -> SearchResult:
"""Search for a query in the collection.

:param str query: Query to search for
:param SearchType search_type: Type of search to perform (optional)
:param IndexType index_type: Type of index to search (optional)
:param int result_threshold: Number of results to return (optional)
:param float score_threshold: Threshold score for the search (optional)
:param float dynamic_score_percentage: Percentage of dynamic score to consider (optional)
:raise SearchError: If the search fails
:return: :class:`SearchResult <SearchResult>` object
:rtype: :class:`videodb.search.SearchResult`
"""
search = SearchFactory(self._connection).get_search(search_type)
return search.search_inside_collection(
collection_id=self.id,
Expand Down Expand Up @@ -161,6 +226,17 @@ def upload(
description: Optional[str] = None,
callback_url: Optional[str] = None,
) -> Union[Video, Audio, Image, None]:
"""Upload a file to the collection.

:param str file_path: Path to the file to be uploaded
:param str url: URL of the file to be uploaded
:param MediaType media_type: MediaType object (optional)
:param str name: Name of the file (optional)
:param str description: Description of the file (optional)
:param str callback_url: URL to receive the callback (optional)
:return: :class:`Video <Video>`, or :class:`Audio <Audio>`, or :class:`Image <Image>` object
:rtype: Union[ :class:`videodb.video.Video`, :class:`videodb.audio.Audio`, :class:`videodb.image.Image`]
"""
upload_data = upload(
self._connection,
file_path,
Expand All @@ -179,12 +255,22 @@ def upload(
return Image(self._connection, **upload_data)

def make_public(self):
"""Make the collection public.

:return: None
:rtype: None
"""
self._connection.patch(
path=f"{ApiPath.collection}/{self.id}", data={"is_public": True}
)
self.is_public = True

def make_private(self):
"""Make the collection private.

:return: None
:rtype: None
"""
self._connection.patch(
path=f"{ApiPath.collection}/{self.id}", data={"is_public": False}
)
Expand Down
Loading