laipeng/reflex
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

File upload to use single syntax (#811)

Browse Source
This commit is contained in:
Elijah Ahianyo 2023-04-13 16:27:02 +00:00 committed by GitHub
parent adf2b8d395
commit 4b7cc6ddf5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 26 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
pynecone/app.py
View File

@ -513,31 +513,26 @@ def upload(app: App):
# get handler function
func = getattr(state, handler.split(".")[-1])
# check if there exists any handler args with annotation UploadFile or List[UploadFile]
# check if there exists any handler args with annotation, List[UploadFile]
for k, v in inspect.getfullargspec(
func.fn if isinstance(func, EventHandler) else func
).annotations.items():
if (
types.is_generic_alias(v)
and types._issubclass(v.__args__[0], UploadFile)
or types._issubclass(v, UploadFile)
if types.is_generic_alias(v) and types._issubclass(
v.__args__[0], UploadFile
):
handler_upload_param = (k, v)
break
if not handler_upload_param:
raise ValueError(
f"`{handler}` handler should have a parameter annotated with one of the following: List["
f"pc.UploadFile], pc.UploadFile "
f"`{handler}` handler should have a parameter annotated as List["
f"pc.UploadFile]"
)
# check if handler supports multi-upload
multi_upload = types._issubclass(handler_upload_param[1], List)
event = Event(
token=token,
name=handler,
payload={handler_upload_param[0]: files if multi_upload else files[0]},
payload={handler_upload_param[0]: files},
)
update = await state.process(event)
return update

51
tests/conftest.py
View File

@ -141,18 +141,7 @@ def dict_mutation_state():
class UploadState(pc.State):
"""The base state for uploading a file."""
img: str
img_list: List[str]
async def handle_upload1(self, file: pc.UploadFile):
"""Handle the upload of a file.
Args:
file: The uploaded file.
"""
pass
async def multi_handle_upload(self, files: List[pc.UploadFile]):
async def handle_upload1(self, files: List[pc.UploadFile]):
"""Handle the upload of a file.
Args:
@ -172,25 +161,15 @@ class SubUploadState(BaseState):
img: str
async def handle_upload(self, file: pc.UploadFile):
async def handle_upload(self, files: List[pc.UploadFile]):
"""Handle the upload of a file.
Args:
file: The uploaded file.
files: The uploaded files.
"""
pass
@pytest.fixture
def upload_event_spec():
"""Create an event Spec for a base state.
Returns:
Event Spec.
"""
return EventSpec(handler=UploadState.handle_upload1, upload=True) # type: ignore
@pytest.fixture
def upload_sub_state_event_spec():
"""Create an event Spec for a substate.
@ -202,13 +181,13 @@ def upload_sub_state_event_spec():
@pytest.fixture
def multi_upload_event_spec():
def upload_event_spec():
"""Create an event Spec for a multi-upload base state.
Returns:
Event Spec.
"""
return EventSpec(handler=UploadState.multi_handle_upload, upload=True) # type: ignore
return EventSpec(handler=UploadState.handle_upload1, upload=True) # type: ignore
@pytest.fixture
@ -226,24 +205,24 @@ def upload_state(tmp_path):
class FileUploadState(pc.State):
"""The base state for uploading a file."""
img: str
img_list: List[str]
async def handle_upload2(self, file):
async def handle_upload2(self, files):
"""Handle the upload of a file.
Args:
file: The uploaded file.
files: The uploaded files.
"""
upload_data = await file.read()
outfile = f"{tmp_path}/{file.filename}"
for file in files:
upload_data = await file.read()
outfile = f"{tmp_path}/{file.filename}"
# Save the file.
with open(outfile, "wb") as file_object:
file_object.write(upload_data)
# Save the file.
with open(outfile, "wb") as file_object:
file_object.write(upload_data)
# Update the img var.
self.img = file.filename
# Update the img var.
self.img_list.append(file.filename)
async def multi_handle_upload(self, files: List[pc.UploadFile]):
"""Handle the upload of a file.

3
tests/test_app.py
View File

@ -474,6 +474,5 @@ async def test_upload_file_without_annotation(upload_state):
await fn([file1, file2])
assert (
err.value.args[0]
== "`upload_state.handle_upload2` handler should have a parameter annotated with one of the following: "
"List[pc.UploadFile], pc.UploadFile "
== "`upload_state.handle_upload2` handler should have a parameter annotated as List[pc.UploadFile]"
)

22
tests/test_utils.py
View File

@ -287,20 +287,6 @@ def test_issubclass(cls: type, cls_check: type, expected: bool):
assert types._issubclass(cls, cls_check) == expected
def test_format_upload_event(upload_event_spec):
"""Test formatting an upload event spec.
Args:
upload_event_spec: The event spec fixture.
"""
assert (
format.format_upload_event(upload_event_spec)
== "uploadFiles(upload_state, result, setResult, "
'upload_state.files, "upload_state.handle_upload1",'
"UPLOAD)"
)
def test_format_sub_state_event(upload_sub_state_event_spec):
"""Test formatting an upload event spec of substate.
@ -314,15 +300,15 @@ def test_format_sub_state_event(upload_sub_state_event_spec):
)
def test_format_multi_upload_event(multi_upload_event_spec):
def test_format_upload_event(upload_event_spec):
"""Test formatting an upload event spec.
Args:
multi_upload_event_spec: The event spec fixture.
upload_event_spec: The event spec fixture.
"""
assert (
format.format_upload_event(multi_upload_event_spec)
format.format_upload_event(upload_event_spec)
== "uploadFiles(upload_state, result, setResult, "
'upload_state.files, "upload_state.multi_handle_upload",'
'upload_state.files, "upload_state.handle_upload1",'
"UPLOAD)"
)