Add new upload file functionality (without backward compatibility). See documentation for details. Add example for flask with sqlalchemy. Removed ADMIN_DISABLE_CROP_IMAGE setting. Use disableCropImage prop in UploadImage widget instead.

Author: vsdudakov

.gitignore

@@ -130,6 +130,9 @@ venv.bak/
 .dmypy.json
 dmypy.json
 
+# ruff
+.ruff_cache/
+
 # Pyre type checker
 .pyre/
 

README.md

@@ -477,15 +477,14 @@ class UserAdmin(TortoiseModelAdmin):
     list_display_links = ("id", "username")
     list_filter = ("id", "username", "is_superuser", "is_active")
     search_fields = ("username",)
-    formfield_overrides = {  # noqa: RUF012
+    formfield_overrides: tp.Any = {  # noqa: RUF012
         "username": (WidgetType.SlugInput, {"required": True}),
         "password": (WidgetType.PasswordInput, {"passwordModalForm": True}),
         "avatar_url": (
-            WidgetType.Upload,
+            WidgetType.UploadImage,
             {
                 "required": False,
-                # Disable crop image for upload field
-                # "disableCropImage": True,
+                # "disableCropImage": True,  # optional: disable image cropping
             },
         ),
     }
@@ -505,10 +504,9 @@ class UserAdmin(TortoiseModelAdmin):
         user.hash_password = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
         await user.save(update_fields=("hash_password",))
 
-    async def orm_save_upload_field(self, obj: tp.Any, field: str, base64: str) -> None:
-        # convert base64 to bytes, upload to s3/filestorage, get url and save or save base64 as is to db (don't recomment it)
-        setattr(obj, field, base64)
-        await obj.save(update_fields=(field,))
+    async def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"
 
 ```
 
@@ -531,6 +529,8 @@ class UserAdmin(TortoiseModelAdmin):
 
 
 ```python
+import typing as tp
+
 from django.db import models
 
 from fastadmin import DjangoModelAdmin, register
@@ -541,6 +541,7 @@ class User(models.Model):
     hash_password = models.CharField(max_length=255)
     is_superuser = models.BooleanField(default=False)
     is_active = models.BooleanField(default=False)
+    avatar_url = models.ImageField(null=True)
 
     def __str__(self):
         return self.username
@@ -562,6 +563,10 @@ class UserAdmin(DjangoModelAdmin):
             return None
         return obj.id
 
+    def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:  # type: ignore[override]
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"
+
 ```
 
 
@@ -591,7 +596,7 @@ from sqlalchemy import Boolean, Integer, String, Text, select, update
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
-from fastadmin import SqlAlchemyModelAdmin, register
+from fastadmin import SqlAlchemyModelAdmin, WidgetType, register
 
 sqlalchemy_engine = create_async_engine(
     "sqlite+aiosqlite:///:memory:",
@@ -625,6 +630,9 @@ class UserAdmin(SqlAlchemyModelAdmin):
     list_display_links = ("id", "username")
     list_filter = ("id", "username", "is_superuser", "is_active")
     search_fields = ("username",)
+    formfield_overrides = {  # noqa: RUF012
+        "avatar_url": (WidgetType.UploadImage, {"required": False}),
+    }
 
     async def authenticate(self, username: str, password: str) -> uuid.UUID | int | None:
         sessionmaker = self.get_sessionmaker()
@@ -646,13 +654,9 @@ class UserAdmin(SqlAlchemyModelAdmin):
             await session.execute(query)
             await session.commit()
 
-    async def orm_save_upload_field(self, obj: tp.Any, field: str, base64: str) -> None:
-        sessionmaker = self.get_sessionmaker()
-        async with sessionmaker() as session:
-            # convert base64 to bytes, upload to s3/filestorage, get url and save or save base64 as is to db (don't recomment it)
-            query = update(self.model_cls).where(User.id.in_([obj.id])).values(**{field: base64})
-            await session.execute(query)
-            await session.commit()
+    async def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"
 
 ```
 
@@ -681,7 +685,7 @@ import uuid
 import bcrypt
 from pony.orm import Database, LongStr, Optional, PrimaryKey, Required, commit, db_session
 
-from fastadmin import PonyORMModelAdmin, register
+from fastadmin import PonyORMModelAdmin, WidgetType, register
 
 db = Database()
 db.bind(provider="sqlite", filename=":memory:", create_db=True)
@@ -707,6 +711,9 @@ class UserAdmin(PonyORMModelAdmin):
     list_display_links = ("id", "username")
     list_filter = ("id", "username", "is_superuser", "is_active")
     search_fields = ("username",)
+    formfield_overrides = {  # noqa: RUF012
+        "avatar_url": (WidgetType.UploadImage, {"required": False}),
+    }
 
     @db_session
     def authenticate(self, username: str, password: str) -> uuid.UUID | int | None:
@@ -726,14 +733,9 @@ class UserAdmin(PonyORMModelAdmin):
         obj.hash_password = hash_password
         commit()
 
-    @db_session
-    def orm_save_upload_field(self, obj: tp.Any, field: str, base64: str) -> None:
-        obj = next((f for f in self.model_cls.select(id=obj.id)), None)
-        if not obj:
-            return
-        # convert base64 to bytes, upload to s3/filestorage, get url and save or save base64 as is to db (don't recomment it)
-        setattr(obj, field, base64)
-        commit()
+    def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:  # type: ignore[override]
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"
 
 ```
 

docs/build.py

@@ -42,6 +42,14 @@ def read_cls_docstring(cls):
 
 def get_versions():
     return [
+        {
+            "version": "0.4.0",
+            "changes": [
+                "Add new upload file functionality (without backward compatibility). See documentation for details.",
+                "Add example for flask with sqlalchemy",
+                "Removed ADMIN_DISABLE_CROP_IMAGE setting. Use disableCropImage prop in UploadImage widget instead.",
+            ],
+        },
         {
             "version": "0.3.11",
             "changes": [
@@ -676,6 +684,10 @@ async def save_model(self, id, payload):
                     "type": "alert-warning",
                     "content": "See <a href='https://ant.design/components/overview' target='_blank'>antd components</a> for more details (e.g. how they look).",
                 },
+                {
+                    "type": "text",
+                    "content": "For file and image fields use <code>UploadFile</code> and <code>UploadImage</code> widgets in <code>formfield_overrides</code>. Implement <code>upload_file(obj, field_name, file_name, file_content)</code> on the model admin to handle uploads; it must return the file URL (e.g. after saving to disk or S3).",
+                },
                 {
                     "type": "text",
                     "content": "Use <code>formfield_overrides</code> to customize widget props per field. You can set <code>label</code> for a custom field label and <code>help</code> for description text below the field:",

docs/code/models/tortoise.py

@@ -41,11 +41,10 @@ class UserAdmin(TortoiseModelAdmin):
         "username": (WidgetType.SlugInput, {"required": True}),
         "password": (WidgetType.PasswordInput, {"passwordModalForm": True}),
         "avatar_url": (
-            WidgetType.Upload,
+            WidgetType.UploadImage,
             {
                 "required": False,
-                # Disable crop image for upload field
-                # "disableCropImage": True,
+                # "disableCropImage": True,  # optional: disable image cropping
             },
         ),
     }
@@ -78,7 +77,9 @@ async def activate(self, ids: list[int]) -> None:
     async def deactivate(self, ids: list[int]) -> None:
         await self.model_cls.filter(id__in=ids).update(is_active=False)
 
-    async def orm_save_upload_field(self, obj: tp.Any, field: str, base64: str) -> None:
-        # convert base64 to bytes, upload to s3/filestorage, get url and save or save base64 as is to db (don't recomment it)
-        setattr(obj, field, base64)
-        await obj.save(update_fields=(field,))
+    async def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:
+        # save file to media directory or s3/filestorage, then return the file url
+        url = f"/media/{file_name}"
+        setattr(obj, field_name, url)
+        await obj.save(update_fields=(field_name,))
+        return url

docs/code/quick_tutorial/djangoorm.py

@@ -1,3 +1,5 @@
+import typing as tp
+
 from django.db import models
 
 from fastadmin import DjangoModelAdmin, register
@@ -8,6 +10,7 @@ class User(models.Model):
     hash_password = models.CharField(max_length=255)
     is_superuser = models.BooleanField(default=False)
     is_active = models.BooleanField(default=False)
+    avatar_url = models.ImageField(null=True)
 
     def __str__(self):
         return self.username
@@ -28,3 +31,7 @@ def authenticate(self, username, password):
         if not obj.check_password(password):
             return None
         return obj.id
+
+    def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:  # type: ignore[override]
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"

docs/code/quick_tutorial/ponyorm.py

@@ -4,7 +4,7 @@
 import bcrypt
 from pony.orm import Database, LongStr, Optional, PrimaryKey, Required, commit, db_session
 
-from fastadmin import PonyORMModelAdmin, register
+from fastadmin import PonyORMModelAdmin, WidgetType, register
 
 db = Database()
 db.bind(provider="sqlite", filename=":memory:", create_db=True)
@@ -30,6 +30,9 @@ class UserAdmin(PonyORMModelAdmin):
     list_display_links = ("id", "username")
     list_filter = ("id", "username", "is_superuser", "is_active")
     search_fields = ("username",)
+    formfield_overrides = {  # noqa: RUF012
+        "avatar_url": (WidgetType.UploadImage, {"required": False}),
+    }
 
     @db_session
     def authenticate(self, username: str, password: str) -> uuid.UUID | int | None:
@@ -49,11 +52,6 @@ def change_password(self, id: uuid.UUID | int, password: str) -> None:
         obj.hash_password = hash_password
         commit()
 
-    @db_session
-    def orm_save_upload_field(self, obj: tp.Any, field: str, base64: str) -> None:
-        obj = next((f for f in self.model_cls.select(id=obj.id)), None)
-        if not obj:
-            return
-        # convert base64 to bytes, upload to s3/filestorage, get url and save or save base64 as is to db (don't recomment it)
-        setattr(obj, field, base64)
-        commit()
+    def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:  # type: ignore[override]
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"

docs/code/quick_tutorial/sqlalchemy.py

@@ -6,7 +6,7 @@
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
-from fastadmin import SqlAlchemyModelAdmin, register
+from fastadmin import SqlAlchemyModelAdmin, WidgetType, register
 
 sqlalchemy_engine = create_async_engine(
     "sqlite+aiosqlite:///:memory:",
@@ -40,6 +40,9 @@ class UserAdmin(SqlAlchemyModelAdmin):
     list_display_links = ("id", "username")
     list_filter = ("id", "username", "is_superuser", "is_active")
     search_fields = ("username",)
+    formfield_overrides = {  # noqa: RUF012
+        "avatar_url": (WidgetType.UploadImage, {"required": False}),
+    }
 
     async def authenticate(self, username: str, password: str) -> uuid.UUID | int | None:
         sessionmaker = self.get_sessionmaker()
@@ -61,10 +64,6 @@ async def change_password(self, id: uuid.UUID | int | str, password: str) -> Non
             await session.execute(query)
             await session.commit()
 
-    async def orm_save_upload_field(self, obj: tp.Any, field: str, base64: str) -> None:
-        sessionmaker = self.get_sessionmaker()
-        async with sessionmaker() as session:
-            # convert base64 to bytes, upload to s3/filestorage, get url and save or save base64 as is to db (don't recomment it)
-            query = update(self.model_cls).where(User.id.in_([obj.id])).values(**{field: base64})
-            await session.execute(query)
-            await session.commit()
+    async def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"

docs/code/quick_tutorial/tortoise.py

@@ -26,15 +26,14 @@ class UserAdmin(TortoiseModelAdmin):
     list_display_links = ("id", "username")
     list_filter = ("id", "username", "is_superuser", "is_active")
     search_fields = ("username",)
-    formfield_overrides = {  # noqa: RUF012
+    formfield_overrides: tp.Any = {  # noqa: RUF012
         "username": (WidgetType.SlugInput, {"required": True}),
         "password": (WidgetType.PasswordInput, {"passwordModalForm": True}),
         "avatar_url": (
-            WidgetType.Upload,
+            WidgetType.UploadImage,
             {
                 "required": False,
-                # Disable crop image for upload field
-                # "disableCropImage": True,
+                # "disableCropImage": True,  # optional: disable image cropping
             },
         ),
     }
@@ -54,7 +53,6 @@ async def change_password(self, id: UUID | int | str, password: str) -> None:
         user.hash_password = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
         await user.save(update_fields=("hash_password",))
 
-    async def orm_save_upload_field(self, obj: tp.Any, field: str, base64: str) -> None:
-        # convert base64 to bytes, upload to s3/filestorage, get url and save or save base64 as is to db (don't recomment it)
-        setattr(obj, field, base64)
-        await obj.save(update_fields=(field,))
+    async def upload_file(self, obj: tp.Any, field_name: str, file_name: str, file_content: bytes) -> str:
+        # save file to media directory or s3/filestorage, then return the file url
+        return f"/media/{file_name}"