effort to update fastapi/starlette

Signed-off-by: vsoch <[email protected]>

.github/dev-requirements.txt

@@ -1,5 +1,5 @@
 pre-commit
-black
+black==23.3.0
 isort
 flake8
 pytest

.github/workflows/main.yaml

@@ -11,7 +11,7 @@ jobs:
   formatting:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
 
     - name: Setup black linter
       run: conda create --quiet --name black pyflakes

.github/workflows/tests.yaml

@@ -29,7 +29,11 @@ jobs:
     steps:
     - uses: actions/checkout@v3
     - name: Install Dependencies (in case changes)
-      run: pip install -r requirements.txt
+      run: |
+        pip install -r requirements.txt
+        pip install pydantic==1.10.11
+        pip install pydantic-settings
+
     - name: Run tests
       run: |
         # Tests for the API with auth disabled

app/core/config.py

@@ -5,6 +5,7 @@
 import shlex
 import string
 from typing import Optional
+
 from pydantic_settings import BaseSettings
 
 logger = logging.getLogger(__name__)

app/library/auth.py

@@ -51,7 +51,11 @@ def check_auth(
         db, user_name=credentials.username, password=credentials.password
     )
     if not user:
-        raise HTTPException(status_code=401, detail="Incorrect email or password", headers={"WWW-Authenticate": "Basic"})
+        raise HTTPException(
+            status_code=401,
+            detail="Incorrect email or password",
+            headers={"WWW-Authenticate": "Basic"},
+        )
     elif not crud_user.is_active(user):
         raise HTTPException(status_code=400, detail="Inactive user")
     return credentials.username

app/library/flux.py

@@ -72,6 +72,29 @@ def submit_job(handle, fluxjob, user):
     return job
 
 
+def clean_submit_args(kwargs):
+    """
+    Clean up submit arguments
+    """
+    # Clean up Nones
+    cleaned = {}
+    for k, v in kwargs.items():
+        if k == "option_flags" and v is not None:
+            option_flags = {}
+            flags = v.split(",")
+            for flag in flags:
+                if "=" not in flag:
+                    print('Warning: invalid flag {flag} missing "="')
+                    continue
+                option, value = flag.split("=", 1)
+                option_flags[option] = value
+            v = option_flags
+
+        if v is not None:
+            cleaned[k] = v
+    return cleaned
+
+
 def validate_submit_kwargs(kwargs, envars=None, runtime=None):
     """
     Shared function to validate submit, from API or web UI.

app/routers/api.py

@@ -132,16 +132,16 @@ async def jobs_listing(request: Request, user=user_auth):
 
 
 @router.get("/jobs")
-async def list_jobs(request: Request, user=user_auth):
+async def list_jobs(
+    details: bool = False, limit=None, listing: bool = False, user=user_auth
+):
     """
     List flux jobs associated with the handle.
     """
-    try:
-        payload = await request.json()
-    except Exception:
-        payload = {}
+    payload = {"details": details, "limit": limit, "listing": listing}
 
     # Does the requester want details - in dict or listing form?
+    print(payload)
     if helpers.has_boolean_arg(payload, "details"):
         # Job limit (only relevant for details)
         limit = helpers.get_int_arg(payload, "limit")
@@ -182,7 +182,20 @@ async def cancel_job(jobid, user=user_auth):
 
 
 @router.post("/jobs/submit")
-async def submit_job(request: Request, user=user_auth):
+async def submit_job(
+    command=None,
+    num_tasks: int = None,
+    cores_per_task: int = None,
+    gpus_per_task: int = None,
+    num_nodes: int = None,
+    exclusive: bool = False,
+    option_flags: str = None,
+    envars: dict = None,
+    workdir: str = None,
+    runtime: int = None,
+    is_launcher: bool = False,
+    user=user_auth,
+):
     """
     Submit a job to our running cluster.
 
@@ -192,40 +205,28 @@ async def submit_job(request: Request, user=user_auth):
     """
     from app.main import app
 
-    print(f"User for submit is {user}")
-
     # This can bork if no payload is provided
-    try:
-        payload = await request.json()
-    except Exception:
+    if not command:
         return JSONResponse(
             content={"Message": "A 'command' is minimally required."}, status_code=400
         )
 
-    kwargs = {}
-
-    # Required arguments
-    for required in ["command"]:
-        kwargs[required] = payload.get(required)
-
-    # Optional arguments
-    as_int = ["num_tasks", "cores_per_task", "gpus_per_task", "num_nodes"]
-    as_bool = ["exclusive"]
-    as_is = ["option_flags"]
-
-    for optional in as_int + as_bool + as_is:
-        if optional in payload and payload[optional]:
-            if optional in as_bool:
-                kwargs[optional] = bool(payload[optional])
-            elif optional in as_int:
-                kwargs[optional] = int(payload[optional])
-            else:
-                kwargs[optional] = payload[optional]
+    kwargs = {
+        "command": command,
+        "num_tasks": num_tasks,
+        "cores_per_task": cores_per_task,
+        "gpus_per_task": gpus_per_task,
+        "num_nodes": num_nodes,
+        "exclusive": exclusive,
+        "option_flags": option_flags,
+    }
 
     # One off args not provided to JobspecV1
-    envars = payload.get("envars", {})
-    workdir = payload.get("workdir")
-    runtime = payload.get("runtime", 0) or 0
+    envars = envars or {}
+    runtime = runtime or 0
+
+    # Clean up submit arguments
+    kwargs = flux_cli.clean_submit_args(kwargs)
 
     # Validate the payload, return meaningful message if something is off
     invalid_messages = flux_cli.validate_submit_kwargs(
@@ -238,7 +239,6 @@ async def submit_job(request: Request, user=user_auth):
         )
 
     # Are we using a launcher instead?
-    is_launcher = payload.get("is_launcher", False)
     if is_launcher:
         message = launcher.launch(kwargs, workdir=workdir, envars=envars, user=user)
         result = jsonable_encoder({"Message": message, "id": "MANY"})

clients/python/flux_restful_client/main/client.py

@@ -266,6 +266,8 @@ def submit(self, command, **kwargs):
             "envars",
         ]:
             # Assume if it's provided, period, the user wants to set it!
+            if optional == "option_flags":
+                kwargs[optional] = utils.flatten_list(kwargs[optional])
             if optional in kwargs:
                 data[optional] = kwargs[optional]
 

clients/python/flux_restful_client/utils/__init__.py

@@ -15,4 +15,5 @@
     write_json,
     write_yaml,
 )
+from .misc import flatten_list
 from .terminal import confirm_action, get_installdir, run_command

clients/python/flux_restful_client/utils/misc.py

@@ -0,0 +1,14 @@
+import os
+from subprocess import PIPE, STDOUT, Popen
+
+def flatten_list(obj):
+    """
+    Flatten a dict to a list of comma separated strings
+    """
+    result = ""
+    for key, value in obj.items():
+        if not result:
+            result += f"{key}={value}"
+        else:
+            result = f"{result},{key}={value}"
+    return result

docs/getting_started/developer-guide.md

@@ -84,13 +84,19 @@ After cloning the repository, install dependencies:
 ```bash
 $ python -m venv env
 $ source env/bin/activate
-$ pip install -r requirements.txt
 ```
 
 Install requirements (note that you also need Flux Python available, which isn't in these requirements as you cannot install from pip).
 
 ```bash
-$ pip install -r requirements.txt
+pip install -r requirements.txt
+```
+
+Note that there is a bug with fastapi so we need to do some installs outside of the requirements.txt:
+
+```bash
+pip install pydantic==1.10.11
+pip install pydantic-settings
 ```
 
 #### 2. Start Service

docs/requirements.txt

@@ -1,6 +1,6 @@
 sphinx<7.2
 markupsafe
-sphinx-immaterial>=0.11.6 
+sphinx-immaterial>=0.11.6
 numpy
 pydantic
 pandas

requirements.txt

@@ -9,6 +9,8 @@ Markdown==3.3.6
 pytest==6.2.5
 pyaml
 sqlalchemy
+# needed by starlette
+httpx
 # install also provides alembic binary
 alembic
 pydantic