mirror of
https://github.com/beak-insights/felicity-lims.git
synced 2025-02-24 00:42:59 +08:00
113 lines
2.6 KiB
Python
113 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import asyncio
|
|
import subprocess
|
|
|
|
import typer
|
|
from uvicorn import Config, Server
|
|
|
|
from felicity.almigrator import FelicityMigrator
|
|
from felicity.logcong import LOGGING_CONFIG
|
|
|
|
alembic_service = FelicityMigrator()
|
|
app = typer.Typer()
|
|
|
|
|
|
@app.command()
|
|
def runserver(
|
|
host: str = "0.0.0.0",
|
|
port: int = 8000,
|
|
workers: int = 1,
|
|
reload: bool = False,
|
|
colors: bool = True,
|
|
) -> None:
|
|
"""Felicity LIMS Server"""
|
|
config = Config(
|
|
app="felicity.main:felicity",
|
|
host=host,
|
|
port=port,
|
|
workers=workers,
|
|
reload=reload,
|
|
use_colors=colors,
|
|
log_config=LOGGING_CONFIG,
|
|
)
|
|
Server(config).run()
|
|
|
|
|
|
@app.command()
|
|
def gunicorn(
|
|
host: str = "0.0.0.0",
|
|
port: int = 8000,
|
|
workers: int = 1,
|
|
reload: bool = False,
|
|
colors: bool = True,
|
|
) -> None:
|
|
"""Felicity LIMS Server"""
|
|
# Build the command to run gunicorn with uvicorn workers
|
|
command = [
|
|
"gunicorn",
|
|
f"--bind={host}:{port}",
|
|
f"--workers={workers}",
|
|
"--reload" if reload else "",
|
|
"--log-level=info",
|
|
"--access-logfile=-" if colors else "",
|
|
"felicity.main:felicity",
|
|
"--worker-class=uvicorn.workers.UvicornWorker",
|
|
]
|
|
|
|
# Join the command list into a single string and print it for debugging
|
|
cmd_str = " ".join(command)
|
|
print(f"Running command: {cmd_str}")
|
|
|
|
# Execute the command
|
|
subprocess.run(cmd_str, shell=True, check=True)
|
|
|
|
|
|
@app.command()
|
|
def upgrade(
|
|
revision: str = typer.Option("head", help="Target revision to upgrade to"),
|
|
) -> None:
|
|
"""Upgrade to a specified revision."""
|
|
alembic_service.upgrade(revision)
|
|
typer.echo(f"Upgraded to revision: {revision}")
|
|
|
|
|
|
@app.command()
|
|
def downgrade(
|
|
revision: str = typer.Argument(..., help="Target revision to downgrade to"),
|
|
) -> None:
|
|
"""Downgrade to a specified revision."""
|
|
alembic_service.downgrade(revision)
|
|
typer.echo(f"Downgraded to revision: {revision}")
|
|
|
|
|
|
@app.command()
|
|
def revision(
|
|
message: str = typer.Argument(..., help="Message for the new revision"),
|
|
) -> None:
|
|
"""Not working --- no idea why yet
|
|
Create a new Alembic revision with a message."""
|
|
alembic_service.create_revision(message)
|
|
typer.echo(f"Created new revision with message: {message}")
|
|
|
|
|
|
@app.command()
|
|
def current() -> None:
|
|
"""Show the current database revision."""
|
|
alembic_service.current()
|
|
|
|
|
|
@app.command()
|
|
def history() -> None:
|
|
"""Show the revision history."""
|
|
alembic_service.history()
|
|
|
|
|
|
@app.command()
|
|
def updates() -> None:
|
|
"""Check for updates"""
|
|
asyncio.run(alembic_service.check_for_updates())
|
|
|
|
|
|
def main() -> None:
|
|
app()
|