Skip to content

File

Bases: Dict[str, Any]

Base class for file object.

It keeps information on a content related to a specific storage. It is a specialized dictionary that provides also attribute style access, the dictionary parent permits easy encoding/decoding to JSON.

Source code in sqlalchemy_file/base.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class BaseFile(typing.Dict[str, Any]):
    """Base class for file object.

    It keeps information on a content related to a specific storage.
    It is a specialized dictionary that provides also attribute style access,
    the dictionary parent permits easy encoding/decoding to JSON.

    """

    def __getitem__(self, key: str) -> Any:
        return dict.__getitem__(self, key)

    def __getattr__(self, name: str) -> Any:
        try:
            return self[name]
        except KeyError:
            raise AttributeError(name)

    def __setitem__(self, key: str, value: Any) -> None:
        if getattr(self, "_frozen", False):
            raise TypeError("Already saved files are immutable")
        return dict.__setitem__(self, key, value)

    __setattr__ = __setitem__

    def __delattr__(self, name: str) -> None:
        if getattr(self, "_frozen", False):
            raise TypeError("Already saved files are immutable")

        try:
            del self[name]
        except KeyError:
            raise AttributeError(name)

    def __delitem__(self, key: str) -> None:
        if object.__getattribute__(self, "_frozen"):
            raise TypeError("Already saved files are immutable")
        dict.__delitem__(self, key)

    def _freeze(self) -> None:
        object.__setattr__(self, "_frozen", True)

    def _thaw(self) -> None:
        object.__setattr__(self, "_frozen", False)

Bases: BaseFile

Takes a file as content and uploads it to the appropriate storage according to the attached Column and file information into the database as JSON.

Default attributes provided for all File include:

Attributes:

Name Type Description
filename str

This is the name of the uploaded file
file_id str

This is the generated UUID for the uploaded file
upload_storage str

Name of the storage used to save the uploaded file
path str

This is a combination of upload_storage and file_id separated by /. This will be use later to retrieve the file
content_type str

This is the content type of the uploaded file
uploaded_at datetime

This is the upload date in ISO format
url str

CDN url of the uploaded file
file StoredFile

Only available for saved content, internally call StorageManager.get_file() on path and return an instance of StoredFile
Source code in sqlalchemy_file/file.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class File(BaseFile):
    """Takes a file as content and uploads it to the appropriate storage
    according to the attached Column and file information into the
    database as JSON.

    Default attributes provided for all ``File`` include:

    Attributes:
        filename (str):  This is the name of the uploaded file
        file_id:   This is the generated UUID for the uploaded file
        upload_storage:   Name of the storage used to save the uploaded file
        path:            This is a  combination of `upload_storage` and `file_id` separated by
                        `/`. This will be use later to retrieve the file
        content_type:   This is the content type of the uploaded file
        uploaded_at (datetime):    This is the upload date in ISO format
        url (str):            CDN url of the uploaded file
        file:           Only available for saved content, internally call
                      [StorageManager.get_file()][sqlalchemy_file.storage.StorageManager.get_file]
                      on path and return an instance of `StoredFile`
    """

    def __init__(
        self,
        content: Any = None,
        filename: Optional[str] = None,
        content_type: Optional[str] = None,
        content_path: Optional[str] = None,
        **kwargs: Dict[str, Any],
    ) -> None:
        if content is None and content_path is None:
            raise ValueError("Either content or content_path must be specified")
        super().__init__(**kwargs)
        if isinstance(content, dict):
            object.__setattr__(self, "original_content", None)
            object.__setattr__(self, "saved", True)
            self.update(content)
            self._freeze()
        else:
            self.content_path = content_path
            if content_path is not None:
                self.original_content = None
                filename = filename or os.path.basename(content_path)
                size = os.path.getsize(content_path)
            else:
                self.original_content = get_content_from_file_obj(content)
                filename = filename or get_filename_from_fileob(content)
                size = get_content_size_from_fileobj(self.original_content)
            content_type = content_type or get_content_type_from_fileobj(
                content, filename
            )
            self.update(
                {
                    "filename": filename,
                    "content_type": content_type,
                    "size": size,
                    "files": [],
                }
            )
            self._thaw()

    def apply_validators(self, validators: List[Validator], key: str = "") -> None:
        """Apply validators to current file."""
        for validator in validators:
            validator.process(self, key)

    def apply_processors(
        self,
        processors: List[Processor],
        upload_storage: Optional[str] = None,
    ) -> None:
        """Apply processors to current file."""
        for processor in processors:
            processor.process(self, upload_storage)
        self._freeze()

    def save_to_storage(self, upload_storage: Optional[str] = None) -> None:
        """Save current file into provided `upload_storage`."""
        extra = self.get("extra", {})
        extra.update({"content_type": self.content_type})

        metadata = self.get("metadata", None)
        if metadata is not None:
            warnings.warn(
                'metadata attribute is deprecated. Use extra={"meta_data": ...} instead',
                DeprecationWarning,
                stacklevel=1,
            )
            extra.update({"meta_data": metadata})

        if extra.get("meta_data", None) is None:
            extra["meta_data"] = {}

        extra["meta_data"].update(
            {"filename": self.filename, "content_type": self.content_type}
        )
        stored_file = self.store_content(
            self.original_content,
            upload_storage,
            extra=extra,
            headers=self.get("headers", None),
            content_path=self.content_path,
        )
        self["file_id"] = stored_file.name
        self["upload_storage"] = upload_storage
        self["uploaded_at"] = datetime.utcnow().isoformat()
        self["path"] = f"{upload_storage}/{stored_file.name}"
        self["url"] = stored_file.get_cdn_url()
        self["saved"] = True

    def store_content(
        self,
        content: Any,
        upload_storage: Optional[str] = None,
        name: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        extra: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
        content_path: Optional[str] = None,
    ) -> StoredFile:
        """Store content into provided `upload_storage`
        with additional `metadata`. Can be used by processors
        to store additional files.
        """
        name = name or str(uuid.uuid4())
        stored_file = StorageManager.save_file(
            name=name,
            content=content,
            upload_storage=upload_storage,
            metadata=metadata,
            extra=extra,
            headers=headers,
            content_path=content_path,
        )
        self["files"].append(f"{upload_storage}/{name}")
        return stored_file

    def encode(self) -> Dict[str, Any]:
        return {k: v for k, v in self.items() if k not in ["original_content"]}

    @classmethod
    def decode(cls, data: Any) -> "File":
        return cls(data)

    @property
    def file(self) -> "StoredFile":
        if self.get("saved", False):
            return StorageManager.get_file(self["path"])
        raise RuntimeError("Only available for saved file")

apply_processors(processors, upload_storage=None)

Apply processors to current file.

Source code in sqlalchemy_file/file.py 
85
86
87
88
89
90
91
92
93
def apply_processors(
    self,
    processors: List[Processor],
    upload_storage: Optional[str] = None,
) -> None:
    """Apply processors to current file."""
    for processor in processors:
        processor.process(self, upload_storage)
    self._freeze()

apply_validators(validators, key='')

Apply validators to current file.

Source code in sqlalchemy_file/file.py 
80
81
82
83
def apply_validators(self, validators: List[Validator], key: str = "") -> None:
    """Apply validators to current file."""
    for validator in validators:
        validator.process(self, key)

save_to_storage(upload_storage=None)

Save current file into provided upload_storage.

Source code in sqlalchemy_file/file.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def save_to_storage(self, upload_storage: Optional[str] = None) -> None:
    """Save current file into provided `upload_storage`."""
    extra = self.get("extra", {})
    extra.update({"content_type": self.content_type})

    metadata = self.get("metadata", None)
    if metadata is not None:
        warnings.warn(
            'metadata attribute is deprecated. Use extra={"meta_data": ...} instead',
            DeprecationWarning,
            stacklevel=1,
        )
        extra.update({"meta_data": metadata})

    if extra.get("meta_data", None) is None:
        extra["meta_data"] = {}

    extra["meta_data"].update(
        {"filename": self.filename, "content_type": self.content_type}
    )
    stored_file = self.store_content(
        self.original_content,
        upload_storage,
        extra=extra,
        headers=self.get("headers", None),
        content_path=self.content_path,
    )
    self["file_id"] = stored_file.name
    self["upload_storage"] = upload_storage
    self["uploaded_at"] = datetime.utcnow().isoformat()
    self["path"] = f"{upload_storage}/{stored_file.name}"
    self["url"] = stored_file.get_cdn_url()
    self["saved"] = True

store_content(content, upload_storage=None, name=None, metadata=None, extra=None, headers=None, content_path=None)

Store content into provided upload_storage with additional metadata. Can be used by processors to store additional files.

Source code in sqlalchemy_file/file.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def store_content(
    self,
    content: Any,
    upload_storage: Optional[str] = None,
    name: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    extra: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    content_path: Optional[str] = None,
) -> StoredFile:
    """Store content into provided `upload_storage`
    with additional `metadata`. Can be used by processors
    to store additional files.
    """
    name = name or str(uuid.uuid4())
    stored_file = StorageManager.save_file(
        name=name,
        content=content,
        upload_storage=upload_storage,
        metadata=metadata,
        extra=extra,
        headers=headers,
        content_path=content_path,
    )
    self["files"].append(f"{upload_storage}/{name}")
    return stored_file