TimelessP/fastapiexample.py

## fastapiexample.py
import time
import fastapi
from fastapi.responses import HTMLResponse
import logging
import uvicorn
from fastapi_utils.tasks import repeat_every

# requirements.txt:
# fastapi~=0.65.1
# uvicorn~=0.13.4
# fastapi-utils~=0.2.1

# FastAPI documentation about repeated-tasks:
# https://fastapi-utils.davidmontague.xyz/user-guide/repeated-tasks/

logging.basicConfig(level=logging.INFO)
api_state = {
    "counter": 0
}
api_config = {
    "simulator_interval_seconds": 0.1
}
api = fastapi.FastAPI()


@api.get("/", response_class=HTMLResponse)
async def index_page():
    html_content = """
        <html>
            <head>
                <title>Some HTML in here</title>
            </head>
            <body>
                <h1>Look ma! HTML!</h1>
            </body>
        </html>
        """
    return HTMLResponse(content=html_content, status_code=200)


@api.get("/api/v1/server/state")
def server_state():
    api_state["counter"] = (api_state["counter"] + 1) % 10
    logging.info(f"tock {api_state['counter']=}")
    return {"counter": api_state["counter"]}


@api.get("/api/v1/worker/sleep/{seconds}")
def api_sleep(seconds: float):
    time.sleep(seconds)  # Doesn't block other client requests whilst it does this.
    return {"message": f"I finished sleeping for {seconds=}."}


@api.on_event("startup")
@repeat_every(seconds=1.0)
def simulator():  # This function MUST NOT take arguments
    time.sleep(10.0)  # Does not re-enter this function until it's done.
    api_state["counter"] = (api_state["counter"] + 1) % 10
    logging.info(f"tick {api_state['counter']=}")


if __name__ == "__main__":
    uvicorn.run(api, host="0.0.0.0", port=8394)
	import time
	import fastapi
	from fastapi.responses import HTMLResponse
	import logging
	import uvicorn
	from fastapi_utils.tasks import repeat_every

	# requirements.txt:
	# fastapi~=0.65.1
	# uvicorn~=0.13.4
	# fastapi-utils~=0.2.1

	# FastAPI documentation about repeated-tasks:
	# https://fastapi-utils.davidmontague.xyz/user-guide/repeated-tasks/

	logging.basicConfig(level=logging.INFO)
	api_state = {
	"counter": 0
	}
	api_config = {
	"simulator_interval_seconds": 0.1
	}
	api = fastapi.FastAPI()


	@api.get("/", response_class=HTMLResponse)
	async def index_page():
	html_content = """
	<html>
	<head>
	<title>Some HTML in here</title>
	</head>
	<body>
	<h1>Look ma! HTML!</h1>
	</body>
	</html>
	"""
	return HTMLResponse(content=html_content, status_code=200)


	@api.get("/api/v1/server/state")
	def server_state():
	api_state["counter"] = (api_state["counter"] + 1) % 10
	logging.info(f"tock {api_state['counter']=}")
	return {"counter": api_state["counter"]}


	@api.get("/api/v1/worker/sleep/{seconds}")
	def api_sleep(seconds: float):
	time.sleep(seconds) # Doesn't block other client requests whilst it does this.
	return {"message": f"I finished sleeping for {seconds=}."}


	@api.on_event("startup")
	@repeat_every(seconds=1.0)
	def simulator(): # This function MUST NOT take arguments
	time.sleep(10.0) # Does not re-enter this function until it's done.
	api_state["counter"] = (api_state["counter"] + 1) % 10
	logging.info(f"tick {api_state['counter']=}")


	if __name__ == "__main__":
	uvicorn.run(api, host="0.0.0.0", port=8394)