Spaces:
Running
Running
File size: 1,693 Bytes
b8b73b2 ca01fa3 b8b73b2 121923b ca01fa3 bc2b550 121923b ca01fa3 e7fa7ee ca01fa3 bc2b550 05acf81 bc2b550 05acf81 bc2b550 05acf81 ca01fa3 05acf81 a06b506 05acf81 bc2b550 b8b73b2 05acf81 |
1 2 3 4 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 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
import dataclasses
import fastapi
import pathlib
import pkgutil
from . import ops
from . import workspace
here = pathlib.Path(__file__).parent
for _, name, _ in pkgutil.iter_modules([str(here)]):
if name.endswith('_ops') and not name.startswith('test_'):
print(f'Importing {name}')
__import__(f'server.{name}')
app = fastapi.FastAPI()
@app.get("/api/catalog")
def get_catalog():
return {
k: {op.name: op.model_dump() for op in v.values()}
for k, v in ops.CATALOGS.items()}
class SaveRequest(workspace.BaseConfig):
path: str
ws: workspace.Workspace
def save(req: SaveRequest):
path = DATA_PATH / req.path
assert path.is_relative_to(DATA_PATH)
workspace.save(req.ws, path)
@app.post("/api/save")
def save_and_execute(req: SaveRequest):
save(req)
workspace.execute(req.ws)
save(req)
return req.ws
@app.get("/api/load")
def load(path: str):
path = DATA_PATH / path
assert path.is_relative_to(DATA_PATH)
if not path.exists():
return workspace.Workspace()
return workspace.load(path)
DATA_PATH = pathlib.Path.cwd() / 'data'
@dataclasses.dataclass(order=True)
class DirectoryEntry:
name: str
type: str
@app.get("/api/dir/list")
def list_dir(path: str):
path = DATA_PATH / path
assert path.is_relative_to(DATA_PATH)
return sorted([
DirectoryEntry(p.relative_to(DATA_PATH), 'directory' if p.is_dir() else 'workspace')
for p in path.iterdir()])
@app.post("/api/dir/mkdir")
def make_dir(req: dict):
path = DATA_PATH / req['path']
assert path.is_relative_to(DATA_PATH)
assert not path.exists()
path.mkdir()
return list_dir(path.parent)
|