Merge pull request #2 from experimental-design/feature/jobber

Jobber based Candidate Generation

.gitignore

@@ -1,3 +1,5 @@
+db.json
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

README.md

@@ -4,13 +4,15 @@
 [![Lint](https://github.com/experimental-design/bofire-candidates-api/workflows/Lint/badge.svg)](https://github.com/experimental-design/bofire-candidates-api/actions?query=workflow%3ALint)
 
 
-An **experimental** FastAPI based application that can be used to generate candidates via https using BoFire. It makes use of the pydantic `data_models` in BoFire, which allows for an easy fastAPI integration include full Swagger documentation which can be found when visiting `/docs` of the running web application.
+An **experimental** FastAPI based application that can be used to generate candidates via http(s) using BoFire. It makes use of pydantic `data_models` in BoFire, which allows for an easy fastAPI integration including Swagger based documentation which can be found when visiting `/docs` of the running web application.
 
-Currently candidates are generated directly at request which can lead to http timeouts and other problems. Future versions should include an asynchronous worker based scenario for generating candidates.
+Candidates can be generated via two different ways, either directly at request which can lead to http timeouts or using an asynchronous worker based procedure.
 
 ## Usage
 
-The following snippet shows how to use the candidates API via Pythons `request` module.
+### Direct Candidate Generation
+
+In the following it is shown how to generate candidates in the direct way using a post request.
 
 ```python
 import json
@@ -78,6 +80,35 @@ df_candidates =Candidates(**json.loads(response.content)).to_pandas()
 
 ```
 
+### Worker Based Candidate Generation
+
+The following snippet shows how to use the worker based candidate generation using the same payload as above. The API is storing all necessary information regarding the proposals in a [`TinyDB`](https://tinydb.readthedocs.io/en/latest/) database. **Note that concurrent worker access using multiple users has not been tested yet.**
+
+``` python
+import time
+
+# create the proposal in the database 
+response = requests.post(url=f"{URL}/proposals", json=payload.model_dump(), headers=HEADERS)
+id = json.loads(response.content)["id"]
+
+# poll the state of the proposal 
+def get_state(id:int):
+    return requests.get(url=f"{URL}proposals/{id}/state", headers=HEADERS).json()
+
+state = get_state(id)
+
+while state in ["CREATED", "CLAIMED"]:
+    state = get_state(id)
+    time.sleep(5)
+
+# get the candidates when the worker is finished
+if state=="FINISHED":
+    response = requests.get(url=f"{URL}proposals/{id}/candidates", headers=HEADERS)
+    candidates = Candidates(**response.json()).to_pandas()
+else:
+    print(state) # candidate generation was not successful.
+```
+
 
 ## Installation
 
@@ -95,6 +126,13 @@ pip install -r requirements.txt
 uvicorn --app-dir app:app --reload
 ```
 
+If you also want to use the asynchronous worker based candidate generation, use the following snippet to start at least one worker:
+
+```bash
+python worker
+```
+
+
 ### Run unit tests
 
 ```bash

app/app.py

@@ -1,10 +1,13 @@
+import bofire
 from fastapi import FastAPI
 from routers.candidates import router as candidates_router
-from routers.versions import router as versions_router
+from routers.proposals import router as proposals_router
 from starlette.responses import RedirectResponse
 
 
-app = FastAPI(title="BoFire Candidates API", version="0.1.0", root_path="/")
+APP_VERSION = "0.0.1"
+
+app = FastAPI(title="BoFire Candidates API", version=APP_VERSION, root_path="/")
 
 
 @app.get("/", include_in_schema=False)
@@ -13,4 +16,9 @@ async def redirect():
 
 
 app.include_router(candidates_router)
-app.include_router(versions_router)
+app.include_router(proposals_router)
+
+
+@app.get("/versions", response_model=dict[str, str])
+def get_versions() -> dict[str, str]:
+    return {"bofire_candidates_api": APP_VERSION, "bofire": bofire.__version__}

app/models/candidates.py

@@ -1,20 +1,32 @@
 from typing import Optional
 
+from bofire.data_models.base import BaseModel
 from bofire.data_models.dataframes.api import Candidates, Experiments
 from bofire.data_models.strategies.api import AnyStrategy
-from pydantic import BaseModel, Field, model_validator
+from pydantic import Field, model_validator
 
 
 class CandidateRequest(BaseModel):
-    strategy_data: AnyStrategy
+    """Request model for generating candidates."""
+
+    strategy_data: AnyStrategy = Field(description="BoFire strategy data")
     n_candidates: int = Field(
         default=1, gt=0, description="Number of candidates to generate"
     )
-    experiments: Optional[Experiments]
-    pendings: Optional[Candidates]
+    experiments: Optional[Experiments] = Field(
+        default=None, description="Experiments to provide to the strategy"
+    )
+    pendings: Optional[Candidates] = Field(
+        default=None, description="Candidates that are pending to be executed"
+    )
 
     @model_validator(mode="after")
     def validate_experiments(self):
+        """Validate experiments and pendings against the strategy domain.
+
+        Returns:
+            CandidateRequest: The validated request
+        """
         if self.experiments is not None:
             self.strategy_data.domain.validate_experiments(self.experiments.to_pandas())
         if self.pendings is not None:

app/models/proposals.py

@@ -0,0 +1,83 @@
+import datetime
+from enum import Enum
+from typing import Optional
+
+from bofire.data_models.base import BaseModel
+from bofire.data_models.dataframes.api import Candidates, Experiments
+from bofire.data_models.strategies.api import AnyStrategy
+from pydantic import Field, model_validator
+
+
+class ProposalRequest(BaseModel):
+    """Request model for generating candidates."""
+
+    strategy_data: AnyStrategy = Field(description="BoFire strategy data")
+    n_candidates: int = Field(
+        default=1, gt=0, description="Number of candidates to generate"
+    )
+    experiments: Optional[Experiments] = Field(
+        default=None, description="Experiments to provide to the strategy"
+    )
+    pendings: Optional[Candidates] = Field(
+        default=None, description="Candidates that are pending to be executed"
+    )
+
+    @model_validator(mode="after")
+    def validate_experiments(self):
+        """Validates the experiments."""
+        if self.experiments is not None:
+            self.strategy_data.domain.validate_experiments(self.experiments.to_pandas())
+        return self
+
+    @model_validator(mode="after")
+    def validate_pendings(self):
+        """Validates that pendings are None."""
+        if self.pendings is not None:
+            raise ValueError("Pendings must be None for proposals.")
+        return self
+
+
+class StateEnum(str, Enum):
+    """Enum for the state of a proposal."""
+
+    CREATED = "CREATED"
+    CLAIMED = "CLAIMED"
+    FAILED = "FAILED"
+    FINISHED = "FINISHED"
+
+
+class Proposal(ProposalRequest):
+    """Model for a candidates proposal."""
+
+    id: Optional[int] = Field(default=None, description="Proposal ID")
+    candidates: Optional[Candidates] = Field(
+        default=None, description="Candidates generated by the proposal"
+    )
+    created_at: datetime.datetime = Field(
+        default_factory=datetime.datetime.now,
+        description="Timestamp when the proposal was created",
+    )
+    last_updated_at: datetime.datetime = Field(
+        default_factory=datetime.datetime.now,
+        description="Timestamp when the proposal was last updated",
+    )
+    state: StateEnum = Field(
+        default=StateEnum.CREATED, description="State of the proposal"
+    )
+    error_message: Optional[str] = Field(
+        default=None, description="Error message if the proposal failed"
+    )
+
+    @model_validator(mode="after")
+    def validate_candidates(self):
+        """Validates the candidates."""
+        if self.candidates is not None:
+            self.strategy_data.domain.validate_candidates(
+                self.candidates.to_pandas(), only_inputs=True
+            )
+            if len(self.candidates.rows) != self.n_candidates:
+                raise ValueError(
+                    f"Number of candidates ({len(self.candidates.rows)}) does not "
+                    "match n_candidates ({self.n_candidates})."
+                )
+        return self

app/routers/candidates.py

@@ -7,20 +7,45 @@
 router = APIRouter(prefix="", tags=["candidates"])
 
 
+def handle_ask_exceptions(e: Exception) -> None:
+    """Handle exceptions raised by the strategy ask method.
+
+    Args:
+        e (Exception): Exception to handle.
+
+    Raises:
+        HTTPException: Status code 404 if not enough experiments are available to execute the strategy.
+        HTTPException: Status code 500 if other server error occurs.
+    """
+    if str(e) == "Not enough experiments available to execute the strategy.":
+        raise HTTPException(status_code=404, detail=str(e))
+    else:
+        raise HTTPException(
+            status_code=500, detail=f"A server error occurred. Details: {e}"
+        )
+
+
 @router.post("/candidates/generate", response_model=Candidates)
 def generate(
     candidate_request: CandidateRequest,
 ) -> Candidates:
+    """Generate candidates using the specified strategy.
+
+    Args:
+        candidate_request (CandidateRequest): Request model for generating candidates.
+
+    Returns:
+        Candidates: The generated candidates.
+    """
     strategy = strategies.map(candidate_request.strategy_data)
+
     if candidate_request.experiments is not None:
         strategy.tell(candidate_request.experiments.to_pandas())
+
     try:
         df_candidates = strategy.ask(candidate_request.n_candidates)
     except ValueError as e:
-        if str(e) == "Not enough experiments available to execute the strategy.":
-            raise HTTPException(status_code=404, detail=str(e))
-        else:
-            raise HTTPException(
-                status_code=500, detail=f"A server error occurred. Details: {e}"
-            )
+        handle_ask_exceptions(e)
+        pass
+
     return Candidates.from_pandas(df_candidates, candidate_request.strategy_data.domain)