Spaces:
Running
Running
File size: 2,175 Bytes
b8b73b2 ca01fa3 b34d742 b8b73b2 121923b a180fd2 ca01fa3 bc2b550 121923b b34d742 121923b 57ea732 b34d742 ca01fa3 a180fd2 ca01fa3 57ea732 ca01fa3 e7fa7ee 57ea732 ca01fa3 bc2b550 05acf81 bc2b550 05acf81 57ea732 05acf81 bc2b550 05acf81 57ea732 ca01fa3 c9fc495 05acf81 c9fc495 05acf81 57ea732 05acf81 bc2b550 b8b73b2 57ea732 b8b73b2 57ea732 b8b73b2 57ea732 05acf81 57ea732 05acf81 b34d742 57ea732 b34d742 57ea732 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)
|