Malasaur/minecraftd
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Updated responses and endpoint descriptions

Browse source
This commit is contained in:
Malasaur 2025-12-14 15:00:42 +01:00
parent 44340f20b0
commit 17fc911290
No known key found for this signature in database
2 changed files with 169 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

162
main.py
View file

@ -1,5 +1,5 @@
from asyncio import create_task
from typing import Annotated
from typing import Annotated, Optional
import uvicorn
from classes import ProcessStatus
@ -15,42 +15,77 @@ app = FastAPI()
@app.get("/start")
async def start() -> Responses.StartResponse:
"Starts the Server process."
"""Starts the Server's process if it is not already running.
Returns:
status: "started" or "running".
message: The Server's response.
"""
if Controllers.process.status() == ProcessStatus.RUNNING:
return {"status": "running"}
return {
"status": "running",
"message": "The Server was already running.",
}
if Controllers.maintainance.is_set():
Controllers.maintainance.unset()
Controllers.process.start()
return {"status": "started"}
return {
"status": "started",
"message": "The Server was started.",
}
@app.get("/status")
async def status() -> Responses.StatusResponse:
"Checks whether the Server is running and returns its information."
"""Checks whether the Server is running and returns its information.
Returns:
status: One of "online", "offline", "crashed", "maintainance", or "starting".
message: The Server's response.
reason: if status is "maintainance", contains the reason for the Server's maintainance state.
motd: if status is "online", contains the Server's MOTD as an HTML string.
icon: if status is "online", contains the Server's icon as a base64 string.
players: if status is "online", contains:
online: the number of online players.
max: the number of max allowed players.
list: the list of online players' usernames as strings.
"""
process_status = Controllers.process.status()
if process_status != ProcessStatus.RUNNING:
# Crashed
if process_status == ProcessStatus.CRASHED:
return {"status": "crashed"}
return {
"status": "crashed",
"message": "The Server has crashed.",
}
# Maintainance
if Controllers.maintainance.is_set():
return {"status": "maintainance", "reason": Controllers.maintainance.get()}
return {
"status": "maintainance",
"message": "The Server is offline due to maintainance.",
"reason": Controllers.maintainance.get(),
}
# Offline
return {"status": "offline"}
return {
"status": "offline",
"message": "The Server is offline.",
}
server_status = Controllers.server.status()
# Starting
if not server_status["online"]:
return {"status": "starting"}
return {
"status": "starting",
"message": "The Server is starting.",
}
# Online
return {
"status": "online",
"message": "The Server is online.",
"motd": server_status.get("motd", ""),
"icon": server_status.get("icon", None),
"players": server_status.get(
@ -65,15 +100,50 @@ async def status() -> Responses.StatusResponse:
@app.get("/stop")
async def stop(data: Models.StopModel, authorization: Annotated[str, Header()]):
"Stops the Server."
async def stop(
data: Models.StopModel, authorization: Annotated[str, Header()]
) -> Responses.StopResponse:
"""Stops the Server.
It waits for `countdown` seconds, then runs `/stop` on the Server, and kills it after `timeout` seconds if it's still alive.
Args:
countdown: the time in seconds to give the players to leave the Server before initiating the shutdown.
if set to 0, the shutdown phase is started immediately. Defaults to 60.
reason: a brief message explaining why the Server is shutting down. Only shown if countdown is non-zero. Defaults to "".
timeout: the time in seconds to wait before killing the Server if its process hasn't stopped. Defaults to 10.
Headers:
Authorization: the Authorization token.
Returns:
status: "stopping"
message: The Server's response.
"""
check_password(authorization)
create_task(stop_server("STOPPING", data.countdown, data.reason, data.timeout))
return {
"status": "stopping",
"message": "The Server is stopping.",
}
@app.get("/restart")
async def restart(data: Models.RestartModel, authorization: Annotated[str, Header()]):
"Restarts the Server."
async def restart(
data: Models.RestartModel, authorization: Annotated[str, Header()]
) -> Responses.RestartResponse:
"""Restarts the Server.
It waits for `countdown` seconds, then runs `/stop` on the Server, and kills it after `timeout` seconds if it's still alive.
Then, it starts the Server again.
Args:
countdown: the time in seconds to give the players to leave the Server before initiating the shutdown.
if set to 0, the shutdown phase is started immediately. Defaults to 60.
reason: a brief message explaining why the Server is restarting. Only shown if countdown is non-zero. Defaults to "".
timeout: the time in seconds to wait before killing the Server if its process hasn't stopped. Defaults to 10.
Headers:
Authorization: the Authorization token.
Returns:
status: "restarting"
message: The Server's response.
"""
check_password(authorization)
create_task(
stop_server(
@ -84,13 +154,30 @@ async def restart(data: Models.RestartModel, authorization: Annotated[str, Heade
Controllers.process.start,
)
)
return {
"status": "restarting",
"message": "The Server is restarting.",
}
@app.get("/maintainance")
async def maintainance(
data: Models.MaintainanceModel, authorization: Annotated[str, Header()]
):
"Stops the Server and sets it to maintainance status."
) -> Responses.MaintainanceResponse:
"""Stops the Server and sets it to maintainance status.
It waits for `countdown` seconds, then runs `/stop` on the Server, and kills it after `timeout` seconds if it's still alive.
Args:
countdown: the time in seconds to give the players to leave the Server before initiating the shutdown.
if set to 0, the shutdown phase is started immediately. Defaults to 60.
reason: a brief message explaining why the Server is entering maintanance mode. Only shown if countdown is non-zero. Defaults to "".
timeout: the time in seconds to wait before killing the Server if its process hasn't stopped. Defaults to 10.
Headers:
Authorization: the Authorization token.
Returns:
status: "stopping"
message: The Server's response.
"""
check_password(authorization)
create_task(
stop_server(
@ -101,29 +188,64 @@ async def maintainance(
Controllers.maintainance.set(data.reason),
)
)
return {
"status": "stopping",
"message": "The Server is stopping for maintainance.",
}
@app.get("/command")
async def command(
data: Models.CommandModel, authorization: Annotated[str, Header()]
) -> str:
"Runs a command on the Server and returns its output."
) -> Responses.CommandResponse:
"""
Executes a command on the Server and returns its output.
Args:
command: The command to execute.
Headers:
Authorization: the Authorization token.
Returns:
status: "executed"
message: The Server's response.
output: The command's output.
"""
check_password(authorization)
return Controllers.server.command(data.command)
return {
"status": "executed",
"message": "The command was executed.",
"output": Controllers.server.command(data.command),
}
@app.get("/logs")
@app.get("/logs/stream")
async def logs_stream(authorization: Annotated[str, Header()]) -> StreamingResponse:
"""Streams Server logs in real-time using SSE.
Headers:
Authorization: the Authorization token.
Returns: text/event-stream
"""
check_password(authorization)
return StreamingResponse(Controllers.logs.stream(), media_type="text/event-stream")
@app.get("/logs/tail")
async def logs_tail(
data: Models.LogsTailModel, authorization: Annotated[str, Header()]
authorization: Annotated[str, Header()],
data: Optional[Models.LogsTailModel] = None,
) -> StreamingResponse:
"""Streams the last few lines of the Server's logs.
Args:
back: The number of lines to stream.
Headers:
Authorization: the Authorization token.
Returns: text/event-stream
"""
check_password(authorization)
return StreamingResponse(Controllers.logs.tail(data.back))
return StreamingResponse(Controllers.logs.tail(data.back if data else 10))
if __name__ == "__main__":

27
responses.py
View file

@ -5,6 +5,7 @@ from typing_extensions import TypedDict
class StartResponse(TypedDict):
status: Literal["running", "started"]
message: str
class StatusResponsePlayers(TypedDict):
@ -15,13 +16,39 @@ class StatusResponsePlayers(TypedDict):
class StatusResponse(TypedDict):
status: Literal["online", "offline", "crashed", "maintainance", "starting"]
message: str
reason: NotRequired[str]
motd: NotRequired[str]
icon: NotRequired[str | None]
players: NotRequired[StatusResponsePlayers]
class StopResponse(TypedDict):
status: Literal["stopping"]
message: str
class RestartResponse(TypedDict):
status: Literal["restarting"]
message: str
class MaintainanceResponse(TypedDict):
status: Literal["stopping"]
message: str
class CommandResponse(TypedDict):
status: Literal["executed"]
message: str
output: str
class Responses:
StartResponse = StartResponse
StatusResponsePlayers = StatusResponsePlayers
StatusResponse = StatusResponse
StopResponse = StopResponse
RestartResponse = RestartResponse
MaintainanceResponse = MaintainanceResponse
CommandResponse = CommandResponse