From 959fb4bb4ec50af4bc28a4de1455d25ab82b10e8 Mon Sep 17 00:00:00 2001 From: Frederick Muriuki Muriithi Date: Thu, 19 Dec 2024 15:08:10 -0600 Subject: Make files into a module of its own. --- uploader/files/__init__.py | 4 ++++ uploader/files/functions.py | 49 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+) create mode 100644 uploader/files/__init__.py create mode 100644 uploader/files/functions.py (limited to 'uploader/files') diff --git a/uploader/files/__init__.py b/uploader/files/__init__.py new file mode 100644 index 0000000..4a62409 --- /dev/null +++ b/uploader/files/__init__.py @@ -0,0 +1,4 @@ +from .functions import (fullpath, + save_file, + chunked_binary_read, + sha256_digest_over_file) diff --git a/uploader/files/functions.py b/uploader/files/functions.py new file mode 100644 index 0000000..d37a53e --- /dev/null +++ b/uploader/files/functions.py @@ -0,0 +1,49 @@ +"""Utilities to deal with uploaded files.""" +import hashlib +from pathlib import Path +from typing import Iterator +from datetime import datetime + +from flask import current_app + +from werkzeug.utils import secure_filename +from werkzeug.datastructures import FileStorage + +def save_file(fileobj: FileStorage, upload_dir: Path) -> Path: + """Save the uploaded file and return the path.""" + assert bool(fileobj), "Invalid file object!" + hashed_name = hashlib.sha512( + f"{fileobj.filename}::{datetime.now().isoformat()}".encode("utf8") + ).hexdigest() + filename = Path(secure_filename(hashed_name)) # type: ignore[arg-type] + if not upload_dir.exists(): + upload_dir.mkdir() + + filepath = Path(upload_dir, filename) + fileobj.save(filepath) + return filepath + + +def fullpath(filename: str): + """Get a file's full path. This makes use of `flask.current_app`.""" + return Path(current_app.config["UPLOAD_FOLDER"], filename).absolute() + + +def chunked_binary_read(filepath: Path, chunksize: int = 2048) -> Iterator: + """Read a file in binary mode in chunks.""" + with open(filepath, "rb") as inputfile: + while True: + data = inputfile.read(chunksize) + if data != b"": + yield data + continue + break + + +def sha256_digest_over_file(filepath: Path) -> str: + """Compute the sha256 digest over a file's contents.""" + filehash = hashlib.sha256() + for chunk in chunked_binary_read(filepath): + filehash.update(chunk) + + return filehash.hexdigest() -- cgit v1.2.3