Skip to content

gemini_media

delete_uploaded_files 

delete_uploaded_files()

Delete all previously uploaded files from the Gemini API.

Source code in src/prompto/apis/gemini/gemini_media.py 
123
124
125
126
127
128
129
def delete_uploaded_files():
    """
    Delete all previously uploaded files from the Gemini API.
    """
    _init_genai()
    uploaded_files = _get_previously_uploaded_files()
    return asyncio.run(_delete_uploaded_files_async(uploaded_files))

list_uploaded_files 

list_uploaded_files()

List all previously uploaded files to the Gemini API.

Source code in src/prompto/apis/gemini/gemini_media.py 
110
111
112
113
114
115
116
117
118
119
120
def list_uploaded_files():
    """
    List all previously uploaded files to the Gemini API.
    """
    _init_genai()
    uploaded_files = _get_previously_uploaded_files()

    for file_hash, file_name in uploaded_files.items():
        msg = f"File Name: {file_name}, File Hash: {file_hash}"
        logger.info(msg)
    logger.info("All uploaded files listed.")

remote_file_hash_base64 

remote_file_hash_base64(remote_file)

Convert a remote file’s SHA256 hash (stored as a hex-encoded UTF-8 bytes object) to a base64-encoded string.

Source code in src/prompto/apis/gemini/gemini_media.py 
23
24
25
26
27
28
29
30
def remote_file_hash_base64(remote_file):
    """
    Convert a remote file's SHA256 hash (stored as a hex-encoded UTF-8 bytes object)
    to a base64-encoded string.
    """
    hex_str = remote_file.sha256_hash.decode("utf-8")
    raw_bytes = bytes.fromhex(hex_str)
    return base64.b64encode(raw_bytes).decode("utf-8")

upload_media_files 

upload_media_files(files_to_upload: set[str])

Upload media files to the Gemini API.

Parameters:

files_to_upload : set[str] Set of absolute, local, paths of files to upload.

Returns:

dict[str, str] Dictionary mapping local file paths to their corresponding uploaded filenames.

Source code in src/prompto/apis/gemini/gemini_media.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def upload_media_files(files_to_upload: set[str]):
    """
    Upload media files to the Gemini API.

    Parameters:
    ----------
    files_to_upload : set[str]
        Set of absolute, local, paths of files to upload.

    Returns:
    -------
    dict[str, str]
        Dictionary mapping local file paths to their corresponding uploaded filenames.
    """
    _init_genai()
    return asyncio.run(upload_media_files_async(files_to_upload))

upload_single_file async 

upload_single_file(local_file_path, already_uploaded_files)

Upload the file at ‘file_path’ if it hasn’t been uploaded yet. If a file with the same SHA256 (base64-encoded) hash exists, returns its name. Otherwise, uploads the file, waits for it to be processed, and returns the new file’s name. Raises a ValueError if processing fails.

Parameters:

Name Type Description Default
file_path str

Path to the file to be uploaded.

 required
already_uploaded_files dict[str, str]

Dictionary mapping file hashes to filenames of already uploaded files.

 required

Returns:

Type Description
A tuple containing:
  • The remote path and filename of the file.
  • The local path and filename of the file. (This is always the same as local_file_path parameter, but is a convenience for gathering the results later.)
Source code in src/prompto/apis/gemini/gemini_media.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def upload_single_file(local_file_path, already_uploaded_files):
    """
    Upload the file at 'file_path' if it hasn't been uploaded yet.
    If a file with the same SHA256 (base64-encoded) hash exists, returns its name.
    Otherwise, uploads the file, waits for it to be processed,
    and returns the new file's name. Raises a ValueError if processing fails.

    Parameters
    ----------
    file_path : str
        Path to the file to be uploaded.
    already_uploaded_files : dict[str, str]
        Dictionary mapping file hashes to filenames of already uploaded files.
    Returns
    -------
    A tuple containing:
        - The remote path and filename of the file.
        - The local path and filename of the file. (This is always the same as
          `local_file_path` parameter, but is a convenience for gathering the
          results later.)
    """
    local_hash = compute_sha256_base64(local_file_path)

    if local_hash in already_uploaded_files:
        logger.info(
            f"File '{local_file_path}' already uploaded as '{already_uploaded_files[local_hash]}'"
        )
        return already_uploaded_files[local_hash], local_file_path

    # Upload the file if it hasn't been found.
    # Use asyncio.to_thread to run the blocking upload_file function in a separate thread.
    logger.info(f"Uploading {local_file_path} to Gemini API")
    file_obj = await asyncio.to_thread(genai.upload_file, local_file_path)
    file_obj = await wait_for_processing(file_obj)

    if file_obj.state.name == "FAILED":
        err_msg = (
            f"Failure uploaded file '{file_obj.name}'. Error: {file_obj.error_message}"
        )
        raise ValueError(err_msg)
    # logger.info(
    #     f"Uploaded file '{file_obj.name}' with hash '{local_hash}' to Gemini API"
    # )
    already_uploaded_files[local_hash] = file_obj.name
    return file_obj.name, local_file_path

wait_for_processing async 

wait_for_processing(file_obj, poll_interval=1)

Poll until the file is no longer in the ‘PROCESSING’ state. Returns the updated file object.

Source code in src/prompto/apis/gemini/gemini_media.py 
33
34
35
36
37
38
39
40
41
async def wait_for_processing(file_obj, poll_interval=1):
    """
    Poll until the file is no longer in the 'PROCESSING' state.
    Returns the updated file object.
    """
    while file_obj.state.name == "PROCESSING":
        await asyncio.sleep(poll_interval)
        file_obj = genai.get_file(file_obj.name)
    return file_obj