Spaces:
Running
Running
File size: 2,175 Bytes
b8b73b2 ca01fa3 b34d742 b8b73b2 121923b a180fd2 ca01fa3 bc2b550 121923b b34d742 121923b d8f90d7 b34d742 ca01fa3 a180fd2 ca01fa3 d8f90d7 ca01fa3 e7fa7ee d8f90d7 ca01fa3 bc2b550 05acf81 bc2b550 05acf81 d8f90d7 05acf81 bc2b550 05acf81 d8f90d7 ca01fa3 a0194e7 05acf81 a0194e7 05acf81 d8f90d7 05acf81 bc2b550 b8b73b2 d8f90d7 b8b73b2 d8f90d7 b8b73b2 d8f90d7 05acf81 d8f90d7 05acf81 b34d742 d8f90d7 b34d742 d8f90d7 b34d742 |
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 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 |
import dataclasses
import fastapi
import importlib
import pathlib
import pkgutil
from . import crdt
from . import ops
from . import workspace
here = pathlib.Path(__file__).parent
lynxkite_modules = {}
for _, name, _ in pkgutil.iter_modules([str(here)]):
if name.endswith("_ops") and not name.startswith("test_"):
print(f"Importing {name}")
name = f"server.{name}"
lynxkite_modules[name] = importlib.import_module(name)
app = fastapi.FastAPI(lifespan=crdt.lifespan)
app.include_router(crdt.router)
@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")
async def save_and_execute(req: SaveRequest):
save(req)
await 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)
@app.post("/api/service")
async def service(req: dict):
"""Executors can provide extra HTTP APIs through the /api/service endpoint."""
module = lynxkite_modules[req["module"]]
return await module.api_service(req)
|