import os import typing import uuid from fastapi import UploadFile from tortoise.fields import TextField from tortoise import ConfigurationError import io from services.io import delete_root_slash, add_fast_api_root, get_filename_from_path, get_or_create_dir, is_binary_file, get_filename, remove_fastapi_root class FileField(TextField): def __init__(self, *, upload_root: str, **kwargs): super().__init__(**kwargs) self.upload_root = delete_root_slash(upload_root) self.upload_root = get_or_create_dir(os.path.join( os.environ.get('FASTAPI_ROOT_URL'), self.upload_root)) def _is_binary(self, file: UploadFile): return isinstance(file, io.BytesIO) def to_db_value(self, value: typing.IO, instance): if not isinstance(value, str): super().validate(value) value.seek(0) is_binary = self._is_binary(value) name = get_filename(value) parent = get_or_create_dir(os.path.join( self.upload_root, instance.id_code)) mode = 'w+' if not is_binary else 'wb+' path = os.path.join(self.upload_root, parent, name) with open(path, mode) as f: f.write(value.read()) return remove_fastapi_root(path) return value def to_python_value(self, value: str): return value """ abs_path = get_abs_path_from_relative_to_root(value) if is_binary_file(abs_path): mode = 'rb' buffer = io.BytesIO() else: mode = 'r' buffer = io.StringIO() buffer.name = get_filename_from_path(value) with open(abs_path, mode) as f: buffer.write(f.read()) buffer.seek(0) return buffer """