Skip to content

Refactor cli #930

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

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Refactor cli #930

Show file tree
Hide file tree
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
1 change: 0 additions & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@ dependencies = [
"pydantic >=2, <3",
"pydantic-settings >=2, <3",
"pyyaml",
"typer",
]

classifiers = [
Expand Down
200 changes: 11 additions & 189 deletions tiled/commandline/_admin.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,193 +1,15 @@
from pathlib import Path
from typing import Optional
from pydantic import BaseModel
from pydantic_settings import CliApp, CliSubCommand

import typer
from tiled.commandline._api_key import APIKeys
from tiled.commandline._database import Database
from tiled.commandline._principal import Principals

from ._utils import get_context, get_profile # noqa E402

admin_app = typer.Typer()
class Admin(BaseModel):
database: CliSubCommand[Database]
api_keys: CliSubCommand[APIKeys]
principals: CliSubCommand[Principals]


@admin_app.command("initialize-database")
def initialize_database(database_uri: str):
"""
Initialize a SQL database for use by Tiled.
"""
import asyncio

from sqlalchemy.ext.asyncio import create_async_engine

from ..alembic_utils import UninitializedDatabase, check_database, stamp_head
from ..authn_database.alembic_constants import (
ALEMBIC_DIR,
ALEMBIC_INI_TEMPLATE_PATH,
)
from ..authn_database.core import (
ALL_REVISIONS,
REQUIRED_REVISION,
initialize_database,
)
from ..utils import ensure_specified_sql_driver

database_uri = ensure_specified_sql_driver(database_uri)

async def do_setup():
engine = create_async_engine(database_uri)
redacted_url = engine.url._replace(password="[redacted]")
try:
await check_database(engine, REQUIRED_REVISION, ALL_REVISIONS)
except UninitializedDatabase:
# Create tables and stamp (alembic) revision.
typer.echo(
f"Database {redacted_url} is new. Creating tables and marking revision {REQUIRED_REVISION}.",
err=True,
)
await initialize_database(engine)
typer.echo("Database initialized.", err=True)
else:
typer.echo(f"Database at {redacted_url} is already initialized.", err=True)
raise typer.Abort()
await engine.dispose()

asyncio.run(do_setup())
stamp_head(ALEMBIC_INI_TEMPLATE_PATH, ALEMBIC_DIR, database_uri)


@admin_app.command("upgrade-database")
def upgrade_database(
database_uri: str,
revision: Optional[str] = typer.Argument(
None,
help="The ID of a revision to upgrade to. By default, upgrade to the latest one.",
),
):
"""
Upgrade the database schema to the latest version.
"""
import asyncio

from sqlalchemy.ext.asyncio import create_async_engine

from ..alembic_utils import get_current_revision, upgrade
from ..authn_database.alembic_constants import (
ALEMBIC_DIR,
ALEMBIC_INI_TEMPLATE_PATH,
)
from ..authn_database.core import ALL_REVISIONS
from ..utils import ensure_specified_sql_driver

database_uri = ensure_specified_sql_driver(database_uri)

async def do_setup():
engine = create_async_engine(database_uri)
redacted_url = engine.url._replace(password="[redacted]")
current_revision = await get_current_revision(engine, ALL_REVISIONS)
await engine.dispose()
if current_revision is None:
# Create tables and stamp (alembic) revision.
typer.echo(
f"Database {redacted_url} has not been initialized. Use `tiled admin initialize-database`.",
err=True,
)
raise typer.Abort()

asyncio.run(do_setup())
upgrade(ALEMBIC_INI_TEMPLATE_PATH, ALEMBIC_DIR, database_uri, revision or "head")


@admin_app.command("downgrade-database")
def downgrade_database(
database_uri: str,
revision: str = typer.Argument(..., help="The ID of a revision to downgrade to."),
):
"""
Upgrade the database schema to the latest version.
"""
import asyncio

from sqlalchemy.ext.asyncio import create_async_engine

from ..alembic_utils import downgrade, get_current_revision
from ..authn_database.alembic_constants import (
ALEMBIC_DIR,
ALEMBIC_INI_TEMPLATE_PATH,
)
from ..authn_database.core import ALL_REVISIONS
from ..utils import ensure_specified_sql_driver

database_uri = ensure_specified_sql_driver(database_uri)

async def do_setup():
engine = create_async_engine(database_uri)
redacted_url = engine.url._replace(password="[redacted]")
current_revision = await get_current_revision(engine, ALL_REVISIONS)
if current_revision is None:
# Create tables and stamp (alembic) revision.
typer.echo(
f"Database {redacted_url} has not been initialized. Use `tiled admin initialize-database`.",
err=True,
)
raise typer.Abort()

asyncio.run(do_setup())
downgrade(ALEMBIC_INI_TEMPLATE_PATH, ALEMBIC_DIR, database_uri, revision)


@admin_app.command("check-config")
def check_config(
config_path: Path = typer.Argument(
None,
help=(
"Path to a config file or directory of config files. "
"If None, check environment variable TILED_CONFIG. "
"If that is unset, try default location ./config.yml."
),
),
):
"Check configuration file for syntax and validation errors."
import os

from ..config import parse_configs

config_path = config_path or os.getenv("TILED_CONFIG", "config.yml")
try:
parse_configs(config_path)
except Exception as err:
typer.echo(str(err), err=True)
raise typer.Exit(1)
typer.echo("No errors found in configuration.")


@admin_app.command("list-principals")
def list_principals(
profile: Optional[str] = typer.Option(
None, help="If you use more than one Tiled server, use this to specify which."
),
page_offset: int = typer.Argument(0),
page_limit: int = typer.Argument(100, help="Max items to show"),
):
"""
List information about all Principals (users or services) that have ever logged in.
"""
import json

context = get_context(profile)
result = context.admin.list_principals(offset=page_offset, limit=page_limit)
typer.echo(json.dumps(result, indent=2))


@admin_app.command("show-principal")
def show_principal(
profile: Optional[str] = typer.Option(
None, help="If you use more than one Tiled server, use this to specify which."
),
uuid: str = typer.Argument(..., help="UUID identifying Principal of interest"),
):
"""
Show information about one Principal (user or service).
"""
import json

context = get_context(profile)
result = context.admin.show_principal(uuid)
typer.echo(json.dumps(result, indent=2))
def cli_cmd(self) -> None:
CliApp.run_subcommand(self)
Loading
Loading