Skip to content

add health method #984

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
helloyongyang merged 1 commit into from
Apr 2, 2026
Merged

add health method #984

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions lightx2v/server/api/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@ def _setup_routes(self):
def redirect_to_docs():
return RedirectResponse(url="/docs")

@self.app.get("/health")
def health_check():
return {"status": "ok"}
Comment on lines +32 to +34
Copy link
Copy Markdown
Contributor

@gemini-code-assist gemini-code-assist bot Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The current health check is 'shallow' and only returns a static 'ok' status. In a distributed system with background workers, it is a best practice to provide a 'deep' health check that reflects the actual state of the service, such as whether the background processing thread is alive and the current status of the task queue. Additionally, using async def is preferred for simple endpoints in FastAPI to avoid the overhead of the external thread pool.

Suggested change
@self.app.get("/health")
def health_check():
return {"status": "ok"}
@self.app.get("/health")
async def health_check():
# Check if the background processing thread is still running
is_alive = self.processing_thread is not None and self.processing_thread.is_alive()
return {
"status": "ok" if is_alive or self.processing_thread is None else "unhealthy",
"worker_alive": is_alive,
**task_manager.get_service_status()
}


api_router = create_api_router()
self.app.include_router(api_router)

Expand Down
Loading