Bases: TypeDecorator
Provides support for storing attachments to SQLAlchemy models.
FileField can be used as a Column type to
store files into the model. The actual file itself will be uploaded to a specific
libcloud.storage.base.Container
, and only the File
information will be stored on the database as JSON.
FileField is transaction aware, so it will delete
every uploaded file whenever the transaction is rolled back and will
delete any old file whenever the transaction is committed.
You can save str
, bytes
or any python file
object
Each file will be validated by provided validators before being saved into
associate storage libcloud.storage.base.Container
and can go through different
processors before being saved in the database.
Source code in sqlalchemy_file/types.py
| class FileField(types.TypeDecorator): # type: ignore
"""Provides support for storing attachments to **SQLAlchemy** models.
[FileField][sqlalchemy_file.types.FileField] can be used as a Column type to
store files into the model. The actual file itself will be uploaded to a specific
`libcloud.storage.base.Container`, and only the [File][sqlalchemy_file.file.File]
information will be stored on the database as JSON.
[FileField][sqlalchemy_file.types.FileField] is transaction aware, so it will delete
every uploaded file whenever the transaction is rolled back and will
delete any old file whenever the transaction is committed.
You can save `str`, `bytes` or any python `file` object
Each file will be validated by provided validators before being saved into
associate storage `libcloud.storage.base.Container` and can go through different
processors before being saved in the database.
"""
impl = types.JSON
cache_ok = False
@property
def python_type(self) -> Type[Union[File, List[File]]]:
if self.multiple:
return MutableList[File]
return File
def __init__(
self,
*args: Tuple[Any],
upload_storage: Optional[str] = None,
validators: Optional[List[Validator]] = None,
processors: Optional[List[Processor]] = None,
upload_type: Type[File] = File,
multiple: Optional[bool] = False,
extra: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
**kwargs: Dict[str, Any],
) -> None:
"""Parameters:
upload_storage: storage to use
validators: List of validators to apply
processors: List of processors to apply
upload_type: File class to use, could be
used to set custom File class
multiple: Use this to save multiple files
extra: Extra attributes (driver specific)
headers: Additional request headers,
such as CORS headers. For example:
headers = {'Access-Control-Allow-Origin': 'http://mozilla.com'}.
"""
super().__init__(*args, **kwargs)
if processors is None:
processors = []
if validators is None:
validators = []
self.upload_storage = upload_storage
self.upload_type = upload_type
self.multiple = multiple
self.extra = extra
self.headers = headers
self.validators = validators
self.processors = processors
def process_bind_param(
self, value: Any, dialect: Dialect
) -> Union[None, Dict[str, Any], List[Dict[str, Any]]]:
if not value:
return None
if not self.multiple and not isinstance(
value, self.upload_type
): # pragma: no cover
raise ValueError(f"Expected {self.upload_type}, received: {type(value)}")
if self.multiple and not (
isinstance(value, list)
and all(isinstance(v, self.upload_type) for v in value)
): # pragma: no cover
raise ValueError(
f"Expected MutableList[{self.upload_type}], received: {type(value)}"
)
return [v.encode() for v in value] if self.multiple else value.encode()
def process_result_value(
self, value: Any, dialect: Dialect
) -> Union[None, MutableList[File], File]:
if value is None:
return None
if isinstance(value, dict):
return (
MutableList([self.upload_type.decode(value)])
if self.multiple
else self.upload_type.decode(value)
)
return MutableList([self.upload_type.decode(v) for v in value])
|
__init__(*args, upload_storage=None, validators=None, processors=None, upload_type=File, multiple=False, extra=None, headers=None, **kwargs)
Parameters:
upload_storage: storage to use
validators: List of validators to apply
processors: List of processors to apply
upload_type: File class to use, could be
used to set custom File class
multiple: Use this to save multiple files
extra: Extra attributes (driver specific)
headers: Additional request headers,
such as CORS headers. For example:
headers = {'Access-Control-Allow-Origin': 'http://mozilla.com'}.
Source code in sqlalchemy_file/types.py
| def __init__(
self,
*args: Tuple[Any],
upload_storage: Optional[str] = None,
validators: Optional[List[Validator]] = None,
processors: Optional[List[Processor]] = None,
upload_type: Type[File] = File,
multiple: Optional[bool] = False,
extra: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
**kwargs: Dict[str, Any],
) -> None:
"""Parameters:
upload_storage: storage to use
validators: List of validators to apply
processors: List of processors to apply
upload_type: File class to use, could be
used to set custom File class
multiple: Use this to save multiple files
extra: Extra attributes (driver specific)
headers: Additional request headers,
such as CORS headers. For example:
headers = {'Access-Control-Allow-Origin': 'http://mozilla.com'}.
"""
super().__init__(*args, **kwargs)
if processors is None:
processors = []
if validators is None:
validators = []
self.upload_storage = upload_storage
self.upload_type = upload_type
self.multiple = multiple
self.extra = extra
self.headers = headers
self.validators = validators
self.processors = processors
|