Upload a file. Args: app: The app to upload the file for. Returns: The upload function.
(app: App)
| 945 | |
| 946 | |
| 947 | def upload(app: App): |
| 948 | """Upload a file. |
| 949 | |
| 950 | Args: |
| 951 | app: The app to upload the file for. |
| 952 | |
| 953 | Returns: |
| 954 | The upload function. |
| 955 | """ |
| 956 | |
| 957 | async def upload_file(request: Request, files: List[UploadFile]): |
| 958 | """Upload a file. |
| 959 | |
| 960 | Args: |
| 961 | request: The FastAPI request object. |
| 962 | files: The file(s) to upload. |
| 963 | |
| 964 | Returns: |
| 965 | StreamingResponse yielding newline-delimited JSON of StateUpdate |
| 966 | emitted by the upload handler. |
| 967 | |
| 968 | Raises: |
| 969 | ValueError: if there are no args with supported annotation. |
| 970 | TypeError: if a background task is used as the handler. |
| 971 | HTTPException: when the request does not include token / handler headers. |
| 972 | """ |
| 973 | token = request.headers.get("nextpy-client-token") |
| 974 | handler = request.headers.get("nextpy-event-handler") |
| 975 | |
| 976 | if not token or not handler: |
| 977 | raise HTTPException( |
| 978 | status_code=400, |
| 979 | detail="Missing nextpy-client-token or nextpy-event-handler header.", |
| 980 | ) |
| 981 | |
| 982 | # Get the state for the session. |
| 983 | state = await app.state_manager.get_state(token) |
| 984 | |
| 985 | # get the current session ID |
| 986 | # get the current state(parent state/substate) |
| 987 | path = handler.split(".")[:-1] |
| 988 | current_state = state.get_substate(path) |
| 989 | handler_upload_param = () |
| 990 | |
| 991 | # get handler function |
| 992 | func = getattr(type(current_state), handler.split(".")[-1]) |
| 993 | |
| 994 | # check if there exists any handler args with annotation, List[UploadFile] |
| 995 | if isinstance(func, EventHandler): |
| 996 | if func.is_background: |
| 997 | raise TypeError( |
| 998 | f"@xt.background is not supported for upload handler `{handler}`.", |
| 999 | ) |
| 1000 | func = func.fn |
| 1001 | if isinstance(func, functools.partial): |
| 1002 | func = func.func |
| 1003 | for k, v in get_type_hints(func).items(): |
| 1004 | if types.is_generic_alias(v) and types._issubclass( |
no outgoing calls