files

1
def get_md5(file: Path):
2
    obj = hashlib.md5()
3
    with file.open("rb") as fp:
4
        while True:
5
            line = fp.read(_DEFAULT_CHUNK_SIZE)
6
            if not line:
7
                break
8
            obj.update(line)
9
    return obj.hexdigest()

1
def upload(url: str, file: Path) -> bool:
2
    with file.open("rb") as data:
3
        return requests.put(url=url, data=data).status_code in (200, 403)

1
@classmethod
2
def from_path(cls, file: t.Union[str, Path], client: Composio, action: str, app: str) -> te.Self:
3
    file = Path(file)
4
    if not file.exists():
5
        raise SDKFileNotFoundError(f'File not found: {file}. Please provide a valid file path.')
6
    if not file.is_file():
7
        raise SDKFileNotFoundError(f'Not a file: {file}. Please provide a valid file path.')
8
    if not os.access(file, os.R_OK):
9
        raise SDKFileNotFoundError(f'File not readable: {file}. Please check the file permissions.')
10
    mimetype = mimetypes.guess(file=file)
11
    s3meta = client.actions.create_file_upload(app=app, action=action, filename=file.name, mimetype=mimetype, md5=get_md5(file=file))
12
    if not s3meta.exists and (not upload(url=s3meta.url, file=file)):
13
        raise ErrorUploadingFile(f'Error uploading file: {file}')
14
    return cls(name=file.name, mimetype=mimetype, s3key=s3meta.key)

1
def download(self, outdir: Path, chunk_size: int=_DEFAULT_CHUNK_SIZE) -> Path:
2
    outfile = outdir / self.name
3
    outdir.mkdir(exist_ok=True, parents=True)
4
    response = requests.get(url=self.s3url, stream=True)
5
    if response.status_code != 200:
6
        raise ErrorDownloadingFile(f'Error downloading file: {self.s3url}')
7
    with outfile.open('wb') as fd:
8
        for chunk in response.iter_content(chunk_size=chunk_size):
9
            fd.write(chunk)
10
    return outfile