moving to fastAPI

Dockerfile

@@ -20,6 +20,6 @@ COPY --chown=user . $HOME/app
 
 #RUN ls -al
 
-#CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
 
-CMD ["python","indexer.py"]
+#CMD ["python","indexer.py"]

main.py

@@ -0,0 +1,45 @@
+import fastapi
+import json
+import uvicorn
+from fastapi import HTTPException
+from fastapi.responses import HTMLResponse
+from fastapi.middleware.cors import CORSMiddleware
+from sse_starlette.sse import EventSourceResponse
+from starlette.responses import StreamingResponse
+from pydantic import BaseModel
+from typing import List, Dict, Any, Generator, Optional
+from server import client
+from types import (
+    Documents,
+    Embeddings,
+    EmbeddingFunction,
+    IDs,
+    Include,
+    Metadatas,
+    Where,
+    WhereDocument,
+    GetResult,
+    QueryResult,
+    CollectionMetadata,
+)
+
+
+app = fastapi.FastAPI(title="ChromaDB")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+api_base="/api/v1"
+bkend=client()
+
+@app.get(api_base+"")
+def heartbeat():
+    return bkend.heartbeat()
+
+
+if __name__ == "__main__":
+  uvicorn.run(app, host="0.0.0.0", port=8000)
+

server.py

@@ -0,0 +1,16 @@
+from chromadb.config import Settings
+import chromadb
+import time
+
+class client():
+    def __init__(self):
+        self.db = chromadb.Client(Settings(
+            chroma_db_impl="duckdb+parquet",
+            persist_directory="/path/to/persist/directory" # Optional, defaults to .chromadb/ in the current directory
+        ))
+
+    def heartbeat(self):
+        return int(time.time_ns())
+
+    def list_collections(self):
+        return self.db.list_collections()

types.py

@@ -0,0 +1,82 @@
+from pydantic import BaseModel
+from typing import Any, Dict, List, Optional
+#from chromadb.api.types import (
+#    CollectionMetadata,
+#    Include,
+#)
+from typing import Union, Literal
+
+#TODO:Update version-Step1::This file is a copy from chromadb/server/fastapi/types.py
+# Copy CollectionMetadata and Include from chromadb.api.types and comment that import
+# Additionally import Union and Literal from typing for "Include to work"
+CollectionMetadata = Dict[Any, Any]
+
+Include = List[
+    Union[
+        Literal["documents"],
+        Literal["embeddings"],
+        Literal["metadatas"],
+        Literal["distances"],
+    ]
+]
+
+
+class AddEmbedding(BaseModel):  # type: ignore
+    # Pydantic doesn't handle Union types cleanly like Embeddings which has
+    # Union[int, float] so we use Any here to ensure data is parsed
+    # to its original type.
+    embeddings: Optional[List[Any]] = None
+    metadatas: Optional[List[Dict[Any, Any]]] = None
+    documents: Optional[List[str]] = None
+    ids: List[str]
+    increment_index: bool = True
+
+
+class UpdateEmbedding(BaseModel):  # type: ignore
+    embeddings: Optional[List[Any]] = None
+    metadatas: Optional[List[Dict[Any, Any]]] = None
+    documents: Optional[List[str]] = None
+    ids: List[str]
+    increment_index: bool = True
+
+
+class QueryEmbedding(BaseModel):  # type: ignore
+    # TODO: Pydantic doesn't bode well with recursive types so we use generic Dicts
+    # for Where and WhereDocument. This is not ideal, but it works for now since
+    # there is a lot of downstream validation.
+    where: Optional[Dict[Any, Any]] = {}
+    where_document: Optional[Dict[Any, Any]] = {}
+    query_embeddings: List[Any]
+    n_results: int = 10
+    include: Include = ["metadatas", "documents", "distances"]
+
+
+class GetEmbedding(BaseModel):  # type: ignore
+    ids: Optional[List[str]] = None
+    where: Optional[Dict[Any, Any]] = None
+    where_document: Optional[Dict[Any, Any]] = None
+    sort: Optional[str] = None
+    limit: Optional[int] = None
+    offset: Optional[int] = None
+    include: Include = ["metadatas", "documents"]
+
+
+class RawSql(BaseModel):  # type: ignore
+    raw_sql: str
+
+
+class DeleteEmbedding(BaseModel):  # type: ignore
+    ids: Optional[List[str]] = None
+    where: Optional[Dict[Any, Any]] = None
+    where_document: Optional[Dict[Any, Any]] = None
+
+
+class CreateCollection(BaseModel):  # type: ignore
+    name: str
+    metadata: Optional[CollectionMetadata] = None
+    get_or_create: bool = False
+
+
+class UpdateCollection(BaseModel):  # type: ignore
+    new_name: Optional[str] = None
+    new_metadata: Optional[CollectionMetadata] = None