seedboxsync

The SeedboxSync main package.

Modules:

__version__ –

All used version in 1 place.
app –

The Flask app module.
cli –

CLI package using Click framework.
core –

SeedboxSync Core package.
taskmanager –

Starter for the SeedboxSync taskmanager.

Functions:

create_app –

Create and configure the SeedboxSync Flask application.

create_app

create_app(injected_config: dict[str, str | bool] | None = None) -> Flask

Create and configure the SeedboxSync Flask application.

Parameters:

injected_config
(dict[str, str] | None, default: None ) –

Optional configuration overrides used by tests.

Returns:

Flask ( Flask ) –

Configured application instance.

Source code in seedboxsync/__init__.py

def create_app(injected_config: dict[str, str | bool] | None = None) -> Flask:
    """
    Create and configure the SeedboxSync Flask application.

    Args:
        injected_config (dict[str, str] | None): Optional configuration overrides
            used by tests.

    Returns:
        Flask: Configured application instance.
    """
    # Create and configure the app
    app = Flask(
        __name__,
        template_folder="front/templates",
        static_folder="front/static",
        instance_relative_config=True,
    )

    # ══════════════════════════════════════════════════════════════════════════════
    # ⚙️  INIT
    # ══════════════════════════════════════════════════════════════════════════════
    # Configure logger for Flask and Click
    logger.configure_logger(app.logger)

    # Load test config
    if injected_config is not None:
        app.config.from_mapping(injected_config)  # load the test config if passed in

    # Initialize the database
    database = Database(app)
    app.extensions["database"] = database

    # Load config
    Config(app)

    # Initialize Babel
    app.config["BABEL_TRANSLATION_DIRECTORIES"] = "front/translations"
    babel.init_app(app, locale_selector=get_locale)

    # Initialize the cache
    cache.init_app(app)

    # Initialize the login manager and OAuth
    login_manager.init_app(app)
    init_oauth2(app)

    # Register jinja filter
    app.jinja_env.filters["slugify"] = slugify

    # Register blueprint and error handler
    app.register_blueprint(bp_auth)
    app.register_blueprint(bp_frontend)
    app.register_blueprint(bp_settings)
    register_api_blueprint(app)
    app.register_error_handler(Exception, __handle_http_exception)  # type: ignore[arg-type]

    # Set up ProxyFix middleware to handle reverse proxy headers
    app.wsgi_app = ProxyFix(  # type: ignore[method-assign]
        app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_port=1
    )

    # ══════════════════════════════════════════════════════════════════════════════
    # ⚙️  FUNCTIONS
    # ══════════════════════════════════════════════════════════════════════════════
    # Routes
    @app.route("/favicon.ico")
    def favicon() -> Response:  # pyright: ignore [reportUnusedFunction]
        """Serve the favicon from the static directory."""
        return send_from_directory(Path(app.root_path, "front/static"), "favicon.png", mimetype="image/png")

    # Before Request
    @app.before_request
    def init_once() -> None:  # pyright: ignore [reportUnusedFunction]
        """Initialize humanize for each request."""
        humanize_i18n.activate(get_locale())

    @app.before_request
    def check_init_error() -> None:  # pyright: ignore [reportUnusedFunction]
        """Display initialization errors as flash messages if any."""
        init_error = app.config.pop("INIT_ERROR", None)
        if init_error:
            flash(init_error, "danger")

    # Context processor
    @app.context_processor
    def inject_formatters() -> dict[str, Callable[[datetime], str]]:  # pyright: ignore [reportUnusedFunction]
        """Inject custom formatters into the template context."""
        return {"format_datetime": format_datetime}

    @app.context_processor
    def inject_globals() -> dict[str, Any]:  # pyright: ignore [reportUnusedFunction]
        """Inject global variables into the template context."""
        return __inject_globals(app)

    # Template global
    @app.template_global()
    def gravatar(email: str) -> str:  # pyright: ignore [reportUnusedFunction]
        """Return the Gravatar image URL for an email address."""
        return __gravatar(email)

    @app.template_global()
    def get_toasted_messages(with_categories: bool = False, category_filter: Iterable[str] = ()) -> list[str] | list[tuple[str, str]]:
        """Like a flash but for toast."""
        return __get_toasted_messages(with_categories, category_filter)

    return app

version

All used version in 1 place.

app

The Flask app module.

cli

CLI package using Click framework.

Modules:

cli –

Cli module.
commands –

Package with all SeedboxSync commands.
context –

Build a context used by Click.

Classes:

Cli –

SeedboxSync command-line interface.
Context –

SeedboxSync Click context.

Functions:

command –

Create a Click command using the SeedboxSync command class.
group –

Create a Click group using the SeedboxSync group class.
pass_context –

Decorate a callback to receive the custom SeedboxSync Click context.

Cli

Cli(*args: Any, **kwargs: Any)


              flowchart TD
              seedboxsync.cli.Cli[Cli]

              

              click seedboxsync.cli.Cli href "" "seedboxsync.cli.Cli"

SeedboxSync command-line interface.

This class customizes Flask's default CLI by: - using the SeedboxSync context implementation; - hiding Flask-specific global options; - automatically loading commands from the commands package; - using custom command and group classes by default.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to FlaskGroup.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to FlaskGroup.

Methods:

command –

Create a command using the custom SeedboxSync command class.
get_command –

Load a command module dynamically.
group –

Create a group using the custom SeedboxSync group class.
invoke –

Invoke the selected command and handle user interruptions gracefully.
list_commands –

Return the list of available SeedboxSync commands.
parse_args –

Parse command-line arguments without Flask's implicit global options.

Source code in seedboxsync/cli/cli.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """
    Initialize the CLI and remove Flask-specific global options.

    Args:
        *args (Any): Positional arguments forwarded to ``FlaskGroup``.
        **kwargs (Any): Keyword arguments forwarded to ``FlaskGroup``.
    """
    super().__init__(*args, **kwargs)

    self.params = [parameter for parameter in self.params if not self._is_hidden_flask_option(parameter)]

command

command(*args: Any, **kwargs: Any) -> Any

Create a command using the custom SeedboxSync command class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated command.

Source code in seedboxsync/cli/cli.py

def command(self, *args: Any, **kwargs: Any) -> Any:
    """
    Create a command using the custom SeedboxSync command class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated command.
    """
    kwargs.setdefault("cls", Command)
    return super().command(*args, **kwargs)

get_command

get_command(ctx: Context, name: str) -> Command | None

Load a command module dynamically.

Parameters:

ctx
(Context) –

The Click context.
name
(str) –

The command name.

Returns:

Command | None –

The loaded Click command, or None if the command cannot be loaded.

Source code in seedboxsync/cli/cli.py

def get_command(self, ctx: click.Context, name: str) -> click.Command | None:
    """
    Load a command module dynamically.

    Args:
        ctx (click.Context): The Click context.
        name (str): The command name.

    Returns:
        The loaded Click command, or ``None`` if the command cannot be loaded.
    """
    try:
        mod = import_module(f"seedboxsync.cli.commands.cmd_{name}")
    except ImportError as e:
        click.echo(e, err=True)
        click.echo(f"Command '{name}' not found.", err=True)
        return None
    return cast(click.Command, mod.cli)

group

group(*args: Any, **kwargs: Any) -> Any

Create a group using the custom SeedboxSync group class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated group.

Source code in seedboxsync/cli/cli.py

def group(self, *args: Any, **kwargs: Any) -> Any:
    """
    Create a group using the custom SeedboxSync group class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated group.
    """
    kwargs.setdefault("cls", Group)
    return super().group(*args, **kwargs)

invoke

invoke(ctx: Context) -> Any

Invoke the selected command and handle user interruptions gracefully.

Parameters:

ctx
(Context) –

The Click context.

Returns:

Any –

The command result.

Source code in seedboxsync/cli/cli.py

def invoke(self, ctx: click.Context) -> Any:
    """
    Invoke the selected command and handle user interruptions gracefully.

    Args:
        ctx (click.Context): The Click context.

    Returns:
        The command result.
    """
    try:
        return super().invoke(ctx)
    except KeyboardInterrupt as exc:
        click.secho("\nInterrupted by user.", fg="yellow")
        raise ctx.exit(130) from exc

list_commands

list_commands(ctx: Context) -> list[str]

Return the list of available SeedboxSync commands.

Parameters:

ctx
(Context) –

The Click context.

Returns:

list[str] –

A sorted list of available command names.

Source code in seedboxsync/cli/cli.py

def list_commands(self, ctx: click.Context) -> list[str]:
    """
    Return the list of available SeedboxSync commands.

    Args:
        ctx (click.Context): The Click context.

    Returns:
        A sorted list of available command names.
    """
    rv = []

    for path in self._CMD_FOLDER.iterdir():
        if path.is_file() and path.suffix == ".py" and path.stem.startswith("cmd_"):
            rv.append(path.stem[4:])

    rv.sort()
    return rv

parse_args

parse_args(ctx: Context, args: list[str]) -> list[str]

Parse command-line arguments without Flask's implicit global options.

Parameters:

ctx
((Context,)) –

The Click context.
args
(list) –

The command-line arguments.

Returns:

list[str] –

The parsed arguments.

Source code in seedboxsync/cli/cli.py

def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
    """
    Parse command-line arguments without Flask's implicit global options.

    Args:
        ctx (click.Context,): The Click context.
        args (list): The command-line arguments.

    Returns:
        The parsed arguments.
    """
    return click.Group.parse_args(self, ctx, args)

Context


              flowchart TD
              seedboxsync.cli.Context[Context]

              

              click seedboxsync.cli.Context href "" "seedboxsync.cli.Context"

SeedboxSync Click context.

Used by:

seedboxsync cli commands
- cmd_health cli
- cmd_search
  - cli
  - downloaded
  - progress
  - uploaded
- cmd_stats
  - by_month
  - by_year
  - cli
  - total
- cmd_sync
  - blackhole
  - seedbox
- cmd_task
- cmd_user
  - add
  - cli
  - delete
  - edit
  - list_user

Methods:

render –

Render tabular data as a Rich table.

Attributes:

app (Flask) –

Return the current Flask application.

app `cached` `property`

app: Flask

Return the current Flask application.

Returns:

Flask ( Flask ) –

The current Flask application.

render

render(data: Iterable[Any], headers: Headers, title: str | None = None) -> str

Render tabular data as a Rich table.

Parameters:

data
(Iterable[Any]) –

Tabular data to render.
headers
(Headers) –

Column headers.
title
(str | None, default: None ) –

Table title.

Returns:

str ( str ) –

The formatted table..

Source code in seedboxsync/cli/context.py

def render(self, data: Iterable[Any], headers: Headers, title: str | None = None) -> str:
    """
    Render tabular data as a Rich table.

    Args:
        data: Tabular data to render.
        headers: Column headers.
        title: Table title.

    Returns:
        str: The formatted table..
    """
    console = Console()
    table = Table(title=title)

    # Set columns from headers
    for key, header in headers.items():
        if isinstance(header, str):
            column_title = header
            table.add_column(column_title)
            continue

        column_title = header.get("title", key)

        table.add_column(
            column_title,
            justify=header.get("justify", "left"),
            style=header.get("style"),
            no_wrap=header.get("no_wrap", False),
            overflow=header.get("overflow", "ellipsis"),
            width=header.get("width"),
            min_width=header.get("min_width"),
            max_width=header.get("max_width"),
            ratio=header.get("ratio"),
        )

    # Set rows from headers and data
    for row in data:
        table.add_row(*(str(row.get(column, "")) for column in headers))

    # Return as string
    with console.capture() as capture:
        console.print(table)
    return capture.get()

command

command(*args: Any, **kwargs: Any) -> Any

Create a Click command using the SeedboxSync command class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated command.

Source code in seedboxsync/cli/cli.py

def command(*args: Any, **kwargs: Any) -> Any:
    """
    Create a Click command using the SeedboxSync command class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated command.
    """
    kwargs.setdefault("cls", Command)
    return click.command(*args, **kwargs)

group

group(*args: Any, **kwargs: Any) -> Any

Create a Click group using the SeedboxSync group class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated group.

Source code in seedboxsync/cli/cli.py

def group(*args: Any, **kwargs: Any) -> Any:
    """
    Create a Click group using the SeedboxSync group class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated group.
    """
    kwargs.setdefault("cls", Group)
    return click.group(*args, **kwargs)

pass_context

pass_context(func: Any) -> Any

Decorate a callback to receive the custom SeedboxSync Click context.

Parameters:

func
(Any) –

The callback to decorate.

Returns:

Any –

The decorated callback.

Source code in seedboxsync/cli/cli.py

def pass_context(func: Any) -> Any:
    """
    Decorate a callback to receive the custom SeedboxSync Click context.

    Args:
        func: The callback to decorate.

    Returns:
        The decorated callback.
    """
    return click.pass_context(func)

cli

Cli module.

Classes:

Cli –

SeedboxSync command-line interface.
Command –

SeedboxSync Click command using the custom context implementation.
Group –

SeedboxSync Click command group using the custom context implementation.

Functions:

check_root_warning –

Check if the script is being run as root and display a warning message.
command –

Create a Click command using the SeedboxSync command class.
group –

Create a Click group using the SeedboxSync group class.
pass_context –

Decorate a callback to receive the custom SeedboxSync Click context.

Cli

Cli(*args: Any, **kwargs: Any)


              flowchart TD
              seedboxsync.cli.cli.Cli[Cli]

              

              click seedboxsync.cli.cli.Cli href "" "seedboxsync.cli.cli.Cli"

SeedboxSync command-line interface.

This class customizes Flask's default CLI by: - using the SeedboxSync context implementation; - hiding Flask-specific global options; - automatically loading commands from the commands package; - using custom command and group classes by default.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to FlaskGroup.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to FlaskGroup.

Methods:

command –

Create a command using the custom SeedboxSync command class.
get_command –

Load a command module dynamically.
group –

Create a group using the custom SeedboxSync group class.
invoke –

Invoke the selected command and handle user interruptions gracefully.
list_commands –

Return the list of available SeedboxSync commands.
parse_args –

Parse command-line arguments without Flask's implicit global options.

Source code in seedboxsync/cli/cli.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """
    Initialize the CLI and remove Flask-specific global options.

    Args:
        *args (Any): Positional arguments forwarded to ``FlaskGroup``.
        **kwargs (Any): Keyword arguments forwarded to ``FlaskGroup``.
    """
    super().__init__(*args, **kwargs)

    self.params = [parameter for parameter in self.params if not self._is_hidden_flask_option(parameter)]

command

command(*args: Any, **kwargs: Any) -> Any

Create a command using the custom SeedboxSync command class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated command.

Source code in seedboxsync/cli/cli.py

def command(self, *args: Any, **kwargs: Any) -> Any:
    """
    Create a command using the custom SeedboxSync command class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated command.
    """
    kwargs.setdefault("cls", Command)
    return super().command(*args, **kwargs)

get_command

get_command(ctx: Context, name: str) -> Command | None

Load a command module dynamically.

Parameters:

ctx
(Context) –

The Click context.
name
(str) –

The command name.

Returns:

Command | None –

The loaded Click command, or None if the command cannot be loaded.

Source code in seedboxsync/cli/cli.py

def get_command(self, ctx: click.Context, name: str) -> click.Command | None:
    """
    Load a command module dynamically.

    Args:
        ctx (click.Context): The Click context.
        name (str): The command name.

    Returns:
        The loaded Click command, or ``None`` if the command cannot be loaded.
    """
    try:
        mod = import_module(f"seedboxsync.cli.commands.cmd_{name}")
    except ImportError as e:
        click.echo(e, err=True)
        click.echo(f"Command '{name}' not found.", err=True)
        return None
    return cast(click.Command, mod.cli)

group

group(*args: Any, **kwargs: Any) -> Any

Create a group using the custom SeedboxSync group class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated group.

Source code in seedboxsync/cli/cli.py

def group(self, *args: Any, **kwargs: Any) -> Any:
    """
    Create a group using the custom SeedboxSync group class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated group.
    """
    kwargs.setdefault("cls", Group)
    return super().group(*args, **kwargs)

invoke

invoke(ctx: Context) -> Any

Invoke the selected command and handle user interruptions gracefully.

Parameters:

ctx
(Context) –

The Click context.

Returns:

Any –

The command result.

Source code in seedboxsync/cli/cli.py

def invoke(self, ctx: click.Context) -> Any:
    """
    Invoke the selected command and handle user interruptions gracefully.

    Args:
        ctx (click.Context): The Click context.

    Returns:
        The command result.
    """
    try:
        return super().invoke(ctx)
    except KeyboardInterrupt as exc:
        click.secho("\nInterrupted by user.", fg="yellow")
        raise ctx.exit(130) from exc

list_commands

list_commands(ctx: Context) -> list[str]

Return the list of available SeedboxSync commands.

Parameters:

ctx
(Context) –

The Click context.

Returns:

list[str] –

A sorted list of available command names.

Source code in seedboxsync/cli/cli.py

def list_commands(self, ctx: click.Context) -> list[str]:
    """
    Return the list of available SeedboxSync commands.

    Args:
        ctx (click.Context): The Click context.

    Returns:
        A sorted list of available command names.
    """
    rv = []

    for path in self._CMD_FOLDER.iterdir():
        if path.is_file() and path.suffix == ".py" and path.stem.startswith("cmd_"):
            rv.append(path.stem[4:])

    rv.sort()
    return rv

parse_args

parse_args(ctx: Context, args: list[str]) -> list[str]

Parse command-line arguments without Flask's implicit global options.

Parameters:

ctx
((Context,)) –

The Click context.
args
(list) –

The command-line arguments.

Returns:

list[str] –

The parsed arguments.

Source code in seedboxsync/cli/cli.py

def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
    """
    Parse command-line arguments without Flask's implicit global options.

    Args:
        ctx (click.Context,): The Click context.
        args (list): The command-line arguments.

    Returns:
        The parsed arguments.
    """
    return click.Group.parse_args(self, ctx, args)

Command


              flowchart TD
              seedboxsync.cli.cli.Command[Command]

              

              click seedboxsync.cli.cli.Command href "" "seedboxsync.cli.cli.Command"

SeedboxSync Click command using the custom context implementation.

Group


              flowchart TD
              seedboxsync.cli.cli.Group[Group]

              

              click seedboxsync.cli.cli.Group href "" "seedboxsync.cli.cli.Group"

SeedboxSync Click command group using the custom context implementation.

Methods:

command –

Create a command using the SeedboxSync command class by default.
group –

Create a group using the SeedboxSync group class by default.

command

command(*args: Any, **kwargs: Any) -> Any

Create a command using the SeedboxSync command class by default.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated command.

Source code in seedboxsync/cli/cli.py

def command(self, *args: Any, **kwargs: Any) -> Any:
    """
    Create a command using the SeedboxSync command class by default.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated command.
    """
    kwargs.setdefault("cls", Command)
    return super().command(*args, **kwargs)

group

group(*args: Any, **kwargs: Any) -> Any

Create a group using the SeedboxSync group class by default.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated group.

Source code in seedboxsync/cli/cli.py

def group(self, *args: Any, **kwargs: Any) -> Any:
    """
    Create a group using the SeedboxSync group class by default.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated group.
    """
    kwargs.setdefault("cls", Group)
    return super().group(*args, **kwargs)

check_root_warning

check_root_warning() -> None

Check if the script is being run as root and display a warning message.

This function checks the effective user ID (UID) on POSIX systems. If the process is executed with root privileges (UID 0), a warning is printed to stderr advising against running as root and providing guidance for Docker container usage.

Source code in seedboxsync/cli/cli.py

def check_root_warning() -> None:
    """
    Check if the script is being run as root and display a warning message.

    This function checks the effective user ID (UID) on POSIX systems.
    If the process is executed with root privileges (UID 0), a warning
    is printed to stderr advising against running as root and providing
    guidance for Docker container usage.
    """
    if hasattr(os, "geteuid") and os.geteuid() == 0:
        click.secho(
            "⚠️  Warning: You are running SeedboxSync as root. This is discouraged. "
            "If you are using Docker, you may have forgotten to pass the --user flag "
            "with the appropriate UID.",
            fg="yellow",
            bold=True,
            err=True,
        )

command

command(*args: Any, **kwargs: Any) -> Any

Create a Click command using the SeedboxSync command class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated command.

Source code in seedboxsync/cli/cli.py

def command(*args: Any, **kwargs: Any) -> Any:
    """
    Create a Click command using the SeedboxSync command class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated command.
    """
    kwargs.setdefault("cls", Command)
    return click.command(*args, **kwargs)

group

group(*args: Any, **kwargs: Any) -> Any

Create a Click group using the SeedboxSync group class.

Parameters:

*args
(Any, default: () ) –

Positional arguments forwarded to Click.
**kwargs
(Any, default: {} ) –

Keyword arguments forwarded to Click.

Returns:

Any –

The decorated group.

Source code in seedboxsync/cli/cli.py

def group(*args: Any, **kwargs: Any) -> Any:
    """
    Create a Click group using the SeedboxSync group class.

    Args:
        *args (Any): Positional arguments forwarded to Click.
        **kwargs (Any): Keyword arguments forwarded to Click.

    Returns:
        The decorated group.
    """
    kwargs.setdefault("cls", Group)
    return click.group(*args, **kwargs)

pass_context

pass_context(func: Any) -> Any

Decorate a callback to receive the custom SeedboxSync Click context.

Parameters:

func
(Any) –

The callback to decorate.

Returns:

Any –

The decorated callback.

Source code in seedboxsync/cli/cli.py

def pass_context(func: Any) -> Any:
    """
    Decorate a callback to receive the custom SeedboxSync Click context.

    Args:
        func: The callback to decorate.

    Returns:
        The decorated callback.
    """
    return click.pass_context(func)

commands

Package with all SeedboxSync commands.

Modules:

cmd_clean –

All commands related to cleaning operations in SeedboxSync.
cmd_health –

All commands related to health checks in SeedboxSync.
cmd_search –

All commands related to search operations in SeedboxSync.
cmd_stats –

All commands related to statistics operations in SeedboxSync.
cmd_sync –

All commands related to synchronization operations in SeedboxSync.
cmd_task –

All commands related to the task queue manager operations.
cmd_user –

All commands related to the users operations & management.

cmd_clean

All commands related to cleaning operations in SeedboxSync.

Functions:

cli –

Empty function for Click sub commands.
downloaded –

Remove a downloaded files by its ID.
progress –

Remove all entries of files currently in download.

cli

cli() -> None

Empty function for Click sub commands.

Source code in seedboxsync/cli/commands/cmd_clean.py

@click.group("clean", help="Cleaning operations.")
def cli() -> None:
    """Empty function for Click sub commands."""

downloaded

downloaded(id: int) -> None

Remove a downloaded files by its ID.

Allows the user to delete a specific downloaded torrent from the database, enabling it to be re-downloaded.

Prints a message indicating whether the torrent was removed or if no matching ID was found.

Parameters:

id
(int) –

The ID of the downloaded torrent to remove.

Source code in seedboxsync/cli/commands/cmd_clean.py

@cli.command("downloaded", help="Remove a downloaded file by ID to enable re-download.")
@click.argument("id", required=True, type=int)
def downloaded(id: int) -> None:  # noqa: A002
    """
    Remove a downloaded files by its ID.

    Allows the user to delete a specific downloaded torrent from
    the database, enabling it to be re-downloaded.

    Prints a message indicating whether the torrent was removed
    or if no matching ID was found.

    Args:
        id (int): The ID of the downloaded torrent to remove.
    """
    count = Download.delete().where(Download.id == id).execute()
    if count == 0:
        click.echo(f"No downloaded file with id {id}")
    else:
        click.echo(f"Torrent with id {id} was removed")

progress

progress() -> None

Remove all entries of files currently in download.

This command deletes all records from the Download table where the download is not yet finished (finished == 0).

Prints the number of deleted entries.

Source code in seedboxsync/cli/commands/cmd_clean.py

@cli.command("progress", help="Clean the list of files currently in download from seedbox.")
def progress() -> None:
    """
    Remove all entries of files currently in download.

    This command deletes all records from the `Download` table where
    the download is not yet finished (`finished == 0`).

    Prints the number of deleted entries.
    """
    count = Download.delete().where(Download.finished == 0).execute()
    click.echo(f"In progress list cleaned. {count} line(s) deleted")

cmd_health

All commands related to health checks in SeedboxSync.

Functions:

cli –

Show the health status of the SeedboxSync CLI and web service.

cli

cli(ctx: Context) -> None

Show the health status of the SeedboxSync CLI and web service.

Parameters:

ctx
(Context) –

The SeedboxSync Click context.

Source code in seedboxsync/cli/commands/cmd_health.py

@click.command("health")
@pass_context
def cli(ctx: Context) -> None:
    """
    Show the health status of the SeedboxSync CLI and web service.

    Args:
        ctx (Context): The SeedboxSync Click context.
    """
    exit_code = 0

    # CLI part
    click.echo(f"Version: {version}")
    click.secho("CLI - OK", fg="green")

    # Task manager part
    heartbeat: dict[str, Any] | None = None
    try:
        heartbeat = typed_peewee_dict(
            TaskStatus.select(
                TaskStatus.key,
                TaskStatus.running,
                TaskStatus.started,
                TaskStatus.finished,
            )
            .where(TaskStatus.key == "heartbeat")
            .dicts()
            .first()
        )
    except TaskStatus.DoesNotExist:  # type: ignore[attr-defined]
        exit_code = 1

    if heartbeat is None:
        click.secho("Task manager - NOK", fg="red")
        exit_code = 2
    elif datetime.now() - heartbeat["finished"] > timedelta(minutes=5):
        click.secho("Task manager - NOK", fg="red")
        exit_code = 3
    else:
        click.secho("Task manager - OK", fg="green")

    # Flask WebUI part
    health_url = utils.get_web_healthcheck_url()
    try:
        with urlopen(health_url, timeout=5) as response:
            if response.status == 200:
                click.secho("WebUI - OK", fg="green")
            else:
                click.secho("WebUI - NOK", fg="red")
                exit_code = 5
    except URLError:
        click.secho("WebUI - NOK", fg="red")
        exit_code = 6

    ctx.exit(exit_code)

cmd_search

All commands related to search operations in SeedboxSync.

Functions:

cli –

Empty function for Click sub commands.
downloaded –

Search for the most recent files downloaded from the seedbox.
progress –

Search for files currently in download from the seedbox.
uploaded –

Search for the most recent torrents uploaded from blackhole.

cli

cli(ctx: Context) -> None

Empty function for Click sub commands.

Source code in seedboxsync/cli/commands/cmd_search.py

@group("search", help="Search operations.")  # type: ignore[untyped-decorator]
@pass_context
def cli(ctx: Context) -> None:
    """Empty function for Click sub commands."""

downloaded

downloaded(ctx: Context, number: int, search: str) -> None

Search for the most recent files downloaded from the seedbox.

Filters downloads by an optional search term and limits the number of results displayed.

Renders a list of download IDs, paths, finished timestamps, and sizes.

Parameters:

ctx
(Context) –

The Click context object.
number
(int) –

The maximum number of torrents to display.
search
(str) –

An optional search term to filter torrent names.

Source code in seedboxsync/cli/commands/cmd_search.py

@cli.command("downloaded", help="Search last files downloaded from seedbox.")  # type: ignore[untyped-decorator]
@click.option("-n", "--number", type=int, default=10, help="Number of torrents to display.")
@click.option("-s", "--search", help="Term to search.")
@pass_context
def downloaded(ctx: Context, number: int, search: str) -> None:
    """
    Search for the most recent files downloaded from the seedbox.

    Filters downloads by an optional search term and limits
    the number of results displayed.

    Renders a list of download IDs, paths, finished timestamps, and sizes.

    Args:
        ctx (Context): The Click context object.
        number (int): The maximum number of torrents to display.
        search (str): An optional search term to filter torrent names.
    """
    # Build "where" expression
    where = (Download.finished != 0) & Download.path.contains(search) if search else Download.finished != 0

    # DB query
    data = (
        Download.select(
            Download.id,
            fn.SUBSTR(Download.path, -100).alias("path"),
            fn.short_datetime(Download.finished),
            fn.humanize(Download.local_size).alias("size"),
        )
        .where(where)
        .limit(number)
        .order_by(Download.finished.desc())
        .dicts()
    )

    click.echo(
        ctx.render(
            reversed(data),
            headers={
                "id": "Id",
                "path": "Path",
                "finished": "Finished",
                "size": "Size",
            },
        )
    )

progress

progress(ctx: Context, number: int, search: str) -> None

Search for files currently in download from the seedbox.

Filters in-progress downloads by an optional search term and limits the number of results displayed.

Calculates local download progress and ETA, and renders a list including ID, path, start time, progress percentage, ETA, and size.

Parameters:

ctx
(Context) –

The Click context object.
number
(int) –

The maximum number of torrents to display.
search
(str) –

An optional search term to filter torrent names.

Source code in seedboxsync/cli/commands/cmd_search.py

@cli.command("progress", help="Search files currently in download from seedbox.")  # type: ignore[untyped-decorator]
@click.option("-n", "--number", type=int, default=10, help="Number of torrents to display.")
@click.option("-s", "--search", help="Term to search.")
@pass_context
def progress(ctx: Context, number: int, search: str) -> None:
    """
    Search for files currently in download from the seedbox.

    Filters in-progress downloads by an optional search term and limits
    the number of results displayed.

    Calculates local download progress and ETA, and renders a list
    including ID, path, start time, progress percentage, ETA, and size.

    Args:
        ctx (Context): The Click context object.
        number (int): The maximum number of torrents to display.
        search (str): An optional search term to filter torrent names.
    """
    # Build "where" expression
    where = (Download.finished == 0) & Download.path.contains(search) if search else Download.finished == 0

    # Calculate columns
    progress_expr = 100.0 * Download.local_size / fn.NULLIF(Download.seedbox_size, 0)
    eta_expr = (fn.STRFTIME("%s", "now", "localtime") - fn.STRFTIME("%s", Download.started)) * (100.0 - progress_expr) / fn.NULLIF(progress_expr, 0)

    # DB query
    data = (
        Download.select(
            Download.id,
            fn.SUBSTR(Download.path, -100).alias("path"),
            fn.short_datetime(Download.started),
            fn.ROUND(progress_expr, 0).cast("INTEGER").concat("%").alias("progress"),
            fn.naturaldelta(eta_expr).alias("eta"),
            fn.humanize(Download.seedbox_size).alias("size"),
        )
        .where(where)
        .limit(number)
        .order_by(Download.started.desc())
        .dicts()
    )

    click.echo(
        ctx.render(
            reversed(data),
            headers={"id": "Id", "path": "Path", "started": "Started", "progress": "Progress", "eta": "ETA", "size": "Size"},
        )
    )

uploaded

uploaded(ctx: Context, number: int, search: str) -> None

Search for the most recent torrents uploaded from blackhole.

Filters torrents by an optional search term and limits the number of results displayed.

Renders a list of torrent IDs, names, and sent timestamps.

Parameters:

ctx
(Context) –

The Click context object.
number
(int) –

The maximum number of torrents to display.
search
(str) –

An optional search term to filter torrent names.

Source code in seedboxsync/cli/commands/cmd_search.py

@cli.command("uploaded", help="Search last torrents uploaded from blackhole.")  # type: ignore[untyped-decorator]
@click.option("-n", "--number", type=int, default=10, help="Number of torrents to display.")
@click.option("-s", "--search", help="Term to search.")
@pass_context
def uploaded(ctx: Context, number: int, search: str) -> None:
    """
    Search for the most recent torrents uploaded from blackhole.

    Filters torrents by an optional search term and limits
    the number of results displayed.

    Renders a list of torrent IDs, names, and sent timestamps.

    Args:
        ctx (Context): The Click context object.
        number (int): The maximum number of torrents to display.
        search (str): An optional search term to filter torrent names.
    """
    # Build "where" expression
    conditions = []
    if search:
        conditions.append(Torrent.name.contains(search))

    # DB query
    query = (
        Torrent.select(
            Torrent.id,
            Torrent.name,
            fn.coalesce(fn.humanize(Torrent.total_size), "").alias("total_size"),
            fn.coalesce(Torrent.total_files, "").alias("total_files"),
            fn.short_datetime(Torrent.sent),
        )
        .limit(number)
        .order_by(Torrent.sent.desc())
    )

    # if "where" expression
    if conditions:
        query = query.where(*conditions)
    data = query.dicts()

    click.echo(
        ctx.render(
            reversed(data),
            headers={"id": "Id", "name": "Name", "total_files": "File(s)", "total_size": "Size", "sent": "Sent datetime"},
        )
    )

cmd_stats

All commands related to statistics operations in SeedboxSync.

Functions:

by_month –

Show statistics aggregated by month.
by_year –

Show statistics aggregated by year.
cli –

Display statistics about completed downloads.
total –

Show total statistics for all completed downloads.

by_month

by_month(ctx: Context) -> None

Show statistics aggregated by month.

Parameters:

ctx
(Context) –

The Click context object.

Source code in seedboxsync/cli/commands/cmd_stats.py

@cli.command("by-month", help="Show statistics aggregated by month.")  # type: ignore[untyped-decorator]
@pass_context
def by_month(ctx: Context) -> None:
    """
    Show statistics aggregated by month.

    Args:
        ctx (Context): The Click context object.
    """
    _stats_by_period(ctx, "month", "Month")

by_year

by_year(ctx: Context) -> None

Show statistics aggregated by year.

Parameters:

ctx
(Context) –

The Click context object.

Source code in seedboxsync/cli/commands/cmd_stats.py

@cli.command("by-year", help="Show statistics aggregated by year.")  # type: ignore[untyped-decorator]
@pass_context
def by_year(ctx: Context) -> None:
    """
    Show statistics aggregated by year.

    Args:
        ctx (Context): The Click context object.
    """
    _stats_by_period(ctx, "year", "Year")

cli

cli(ctx: Context) -> None

Display statistics about completed downloads.

Parameters:

ctx
(Context) –

The Click context object.

Source code in seedboxsync/cli/commands/cmd_stats.py

@group(
    "stats",
    help="Stats operations.",
    invoke_without_command=True,
    no_args_is_help=False,
)  # type: ignore[untyped-decorator]
@pass_context
def cli(ctx: Context) -> None:
    """
    Display statistics about completed downloads.

    Args:
        ctx (Context): The Click context object.
    """
    if ctx.invoked_subcommand is None:
        click.echo(ctx.get_help())
        click.echo()
        ctx.invoke(total)

total

total(ctx: Context) -> None

Show total statistics for all completed downloads.

Displays the total number of files and the total size.

Parameters:

ctx
(Context) –

The Click context object.

Source code in seedboxsync/cli/commands/cmd_stats.py

@cli.command("total", help="Show total statistics")  # type: ignore[untyped-decorator]
@pass_context
def total(ctx: Context) -> None:
    """
    Show total statistics for all completed downloads.

    Displays the total number of files and the total size.

    Args:
        ctx (Context): The Click context object.
    """
    query = Download.select().where(Download.finished != 0)
    total_files = query.count()
    total_size = sum([d.seedbox_size for d in query if d.seedbox_size])

    stats = [
        {
            "files": total_files,
            "total_size": filesize.naturalsize(total_size, True),
        }
    ]

    click.echo(ctx.render(stats, headers={"files": "Nb files", "total_size": "Total size"}))

cmd_sync

All commands related to synchronization operations in SeedboxSync.

Functions:

blackhole –

Perform the blackhole synchronization.
cli –

Empty function for Click sub commands.
seedbox –

Perform synchronization from the seedbox.

blackhole

blackhole(ctx: Context, dry_run: bool, ping: bool) -> None

Perform the blackhole synchronization.

Parameters:

ctx
(Context) –

The Click context object.
dry_run
(bool) –

Whether to perform a dry run.
ping
(bool) –

Whether to ping a service during execution.

Source code in seedboxsync/cli/commands/cmd_sync.py

@cli.command("blackhole", help="Sync torrent from blackhole to seedbox.")  # type: ignore[untyped-decorator]
@click.option(
    "--dry-run",
    help="List only, do not upload or persist files.",
    is_flag=True,
    default=False,
)
@click.option(
    "-p",
    "--ping",
    help="Ping a service (e.g., Healthchecks) during execution.",
    is_flag=True,
    default=False,
)
@pass_context
def blackhole(ctx: Context, dry_run: bool, ping: bool) -> None:
    """
    Perform the blackhole synchronization.

    Args:
        ctx (Context): The Click context object.
        dry_run (bool): Whether to perform a dry run.
        ping (bool): Whether to ping a service during execution.
    """
    try:
        with ctx.app.task_manager.lock_task(BLACKHOLE_LOCK_NAME):
            blackhole_service(dry_run, ping)
    except TaskLockedException:
        ctx.app.logger.debug("Blackhole sync already running")

cli

cli() -> None

Empty function for Click sub commands.

Source code in seedboxsync/cli/commands/cmd_sync.py

@group("sync", help="Run synchronization operations.")  # type: ignore[untyped-decorator]
def cli() -> None:
    """Empty function for Click sub commands."""

seedbox

seedbox(ctx: Context, dry_run: bool, ping: bool, only_store: bool) -> None

Perform synchronization from the seedbox.

Parameters:

ctx
(Context) –

The Click context object.
dry_run
(bool) –

Whether to list files without downloading or persisting them.
ping
(bool) –

Whether to ping the configured monitoring service.
only_store
(bool) –

Whether to record remote files without downloading them.

Source code in seedboxsync/cli/commands/cmd_sync.py

@cli.command("seedbox", help="Sync files from seedbox.")  # type: ignore[untyped-decorator]
@click.option(
    "--dry-run",
    help="List only, do not upload or persist files.",
    is_flag=True,
    default=False,
)
@click.option(
    "-p",
    "--ping",
    help="Ping a service (e.g., Healthchecks) during execution.",
    is_flag=True,
    default=False,
)
@click.option(
    "-o",
    "--only-store",
    help="Store the file list only, no download; useful for already synced seedbox.",
    is_flag=True,
    default=False,
)
@pass_context
def seedbox(ctx: Context, dry_run: bool, ping: bool, only_store: bool) -> None:
    """
    Perform synchronization from the seedbox.

    Args:
        ctx (Context): The Click context object.
        dry_run (bool): Whether to list files without downloading or persisting them.
        ping (bool): Whether to ping the configured monitoring service.
        only_store (bool): Whether to record remote files without downloading them.
    """
    """
    Perform the blackhole synchronization.

    Args:
        ctx (Context): The Click context object.
        dry_run (bool): Whether to perform a dry run.
        ping (bool): Whether to ping a service during execution.
    """
    try:
        with ctx.app.task_manager.lock_task(SEEDBOX_LOCK_NAME):
            seedbox_service(dry_run, ping, only_store)
    except TaskLockedException:
        ctx.app.logger.debug("Seedbox sync already running")

cmd_task

All commands related to the task queue manager operations.

Functions:

cli –

Empty function for Click sub commands.
tasks_flush –

Remove all data from the queue.
tasks_flush_lock –

Flush any locks that may be held.
tasks_list –

List registered tasks.
tasks_pending –

List pending tasks.
tasks_result –

List all_results() tasks.
tasks_sync_blackhole –

Launch asynchrone task sync blackhole.
tasks_sync_seedbox –

Launch asynchrone task sync seedbox.

cli

cli() -> None

Empty function for Click sub commands.

Source code in seedboxsync/cli/commands/cmd_task.py

@group("task", help="Task operations on task queue management.")  # type: ignore[untyped-decorator]
def cli() -> None:
    """Empty function for Click sub commands."""

tasks_flush

tasks_flush(ctx: Context) -> None

Remove all data from the queue.

Source code in seedboxsync/cli/commands/cmd_task.py

@cli.command("flush", help="Remove all data from the queue, schedule, and result store.")  # type: ignore[untyped-decorator]
@pass_context
def tasks_flush(ctx: Context) -> None:
    """Remove all data from the queue."""
    ctx.app.task_manager.flush()
    click.echo("Queue flushed")

tasks_flush_lock

tasks_flush_lock(ctx: Context) -> None

Flush any locks that may be held.

Source code in seedboxsync/cli/commands/cmd_task.py

@cli.command("flush-lock", help="Flush any locks that may be held.")  # type: ignore[untyped-decorator]
@pass_context
def tasks_flush_lock(ctx: Context) -> None:
    """Flush any locks that may be held."""
    ctx.app.task_manager.flush_locks()
    click.echo("Lock flushed")

tasks_list

tasks_list(ctx: Context) -> None

List registered tasks.

Source code in seedboxsync/cli/commands/cmd_task.py

@cli.command("list", help="List registered tasks.")  # type: ignore[untyped-decorator]
@pass_context
def tasks_list(ctx: Context) -> None:
    """List registered tasks."""
    load_task_modules()
    data = [{"class": task} for task in ctx.app.task_manager._registry._registry]

    click.echo(ctx.render(data, headers={"class": "Class"}))

tasks_pending

tasks_pending(ctx: Context) -> None

List pending tasks.

Source code in seedboxsync/cli/commands/cmd_task.py

@cli.command("pending", help="List pending tasks.")  # type: ignore[untyped-decorator]
@pass_context
def tasks_pending(ctx: Context) -> None:
    """List pending tasks."""
    data = []
    for task in ctx.app.task_manager.pending():
        task_str = str(task)
        data.append(
            {
                "name": getattr(task, "name", task_str.split(": ")[0]),
                "task_id": getattr(task, "id", str(task).split(": ")[-1] if ": " in task_str else task_str),
            }
        )

    click.echo(ctx.render(data, headers={"name": "Task Name", "task_id": "Task ID / UUID"}))

tasks_result

tasks_result(ctx: Context) -> None

List all_results() tasks.

Source code in seedboxsync/cli/commands/cmd_task.py

@cli.command("result", help="List results in the result store. Allows determining the currently running tasks.")  # type: ignore[untyped-decorator]
@pass_context
def tasks_result(ctx: Context) -> None:
    """List all_results() tasks."""
    data = [{"key": task} for task in ctx.app.task_manager.all_results()]

    click.echo(ctx.render(data, headers={"key": "Result key"}))

tasks_sync_blackhole

tasks_sync_blackhole(ctx: Context) -> None

Launch asynchrone task sync blackhole.

Source code in seedboxsync/cli/commands/cmd_task.py

@cli.command("sync-blackhole", help="Launch asynchrone task sync blackhole.")  # type: ignore[untyped-decorator]
@pass_context
def tasks_sync_blackhole(ctx: Context) -> None:
    """Launch asynchrone task sync blackhole."""
    with ctx.app.app_context():
        from seedboxsync.core.taskmanager.task.task_sync_blackhole import sync_blackhole

        sync_blackhole()
    click.echo("Task sync blackhole launched in task manager")

tasks_sync_seedbox

tasks_sync_seedbox(ctx: Context) -> None

Launch asynchrone task sync seedbox.

Source code in seedboxsync/cli/commands/cmd_task.py

@cli.command("sync-seedbox", help="Launch asynchrone task sync seedbox.")  # type: ignore[untyped-decorator]
@pass_context
def tasks_sync_seedbox(ctx: Context) -> None:
    """Launch asynchrone task sync seedbox."""
    with ctx.app.app_context():
        from seedboxsync.core.taskmanager.task.task_sync_seedbox import sync_seedbox

        sync_seedbox()
    click.echo("Task sync seedbox launched in task manager")

cmd_user

All commands related to the users operations & management.

Functions:

add –

Create a new user account in the database.
cli –

Empty function for Click sub commands.
delete –

Delete a user by their unique database identifier.
edit –

Update details for an existing user account.
list_user –

List registered users from the database.

add

add(ctx: Context, username: str, email: str, password: str) -> None

Create a new user account in the database.

Accepts account details via command-line options or interactive prompts, hashes the password, and stores the user record.

Parameters:

ctx
(Context) –

The CLI application context.
username
(str) –

Unique username for the account.
email
(str) –

Unique email address.
password
(str) –

Plain-text password to hash and store.

Source code in seedboxsync/cli/commands/cmd_user.py

@cli.command("add", help="Add a new user.")  # type: ignore[untyped-decorator]
@click.option("-u", "--username", prompt="Username", help="Username for the new account.")
@click.option("-e", "--email", prompt="Email address", help="Unique email address.")
@click.option("-p", "--password", prompt=True, hide_input=True, confirmation_prompt=True, help="Password for the user account.")
@pass_context
def add(ctx: Context, username: str, email: str, password: str) -> None:
    """
    Create a new user account in the database.

    Accepts account details via command-line options or interactive prompts,
    hashes the password, and stores the user record.

    Args:
        ctx (Context): The CLI application context.
        username (str): Unique username for the account.
        email (str): Unique email address.
        password (str): Plain-text password to hash and store.
    """
    # Verify if user already exists before attempting insertion
    if User.get_or_none((User.username == username) | (User.email == email)):
        click.secho(
            f"Error: A user with username '{username}' or email '{email}' already exists.",
            fg="red",
            err=True,
        )
        return

    try:
        user = User.create(
            username=username,
            email=email,
            password=generate_password_hash(password),
        )
        click.secho(f"User '{user.username}' (ID: {user.id}) created successfully.", fg="green")
    except Exception as e:
        click.secho(f"Error: Failed to create user: {e}", fg="red", err=True)

cli

cli(ctx: Context) -> None

Empty function for Click sub commands.

Source code in seedboxsync/cli/commands/cmd_user.py

@group("user", help="User operations & management for SeedboxSync frontend.")  # type: ignore[untyped-decorator]
@pass_context
def cli(ctx: Context) -> None:
    """Empty function for Click sub commands."""

delete

delete(ctx: Context, id: int, yes: bool) -> None

Delete a user by their unique database identifier.

Parameters:

ctx
(Context) –

The CLI application context.
id
(int) –

Database identifier of the user to remove.
yes
(bool) –

Skip confirmation prompt if set to True.

Source code in seedboxsync/cli/commands/cmd_user.py

@cli.command("delete", help="Delete a user.")  # type: ignore[untyped-decorator]
@click.option("--id", type=int, required=True, help="ID of the user to delete.")
@click.option("-y", "--yes", is_flag=True, help="Confirm deletion without prompting.")
@pass_context
def delete(ctx: Context, id: int, yes: bool) -> None:  # noqa: A002
    """
    Delete a user by their unique database identifier.

    Args:
        ctx (Context): The CLI application context.
        id (int): Database identifier of the user to remove.
        yes (bool): Skip confirmation prompt if set to True.
    """
    try:
        user = User.get_by_id(id)
    except User.DoesNotExist:  # pyright: ignore [reportAttributeAccessIssue]
        click.secho(f"Error: User with ID {id} does not exist.", fg="red", err=True)
        return

    if not yes and not click.confirm(f"Are you sure you want to delete user '{user.username}' (ID: {user.id})?"):
        click.echo("Operation canceled.")
        return

    try:
        username = user.username
        user.delete_instance()
        click.secho(f"User '{username}' (ID: {id}) deleted successfully.", fg="green")
    except Exception as e:
        click.secho(f"Error: Failed to delete user {id}: {e}", fg="red", err=True)

edit

edit(
    ctx: Context,
    id: int,
    username: str | None,
    email: str | None,
    password: str | None,
) -> None

Update details for an existing user account.

Fetches the user by ID and updates only the provided fields. Password updates are automatically hashed before saving.

Parameters:

ctx
(Context) –

The CLI application context.
id
(int) –

Database identifier of the user to edit.
username
(str | None) –

New username to set.
email
(str | None) –

New email address to set.
password
(str | None) –

New plain-text password to hash and update.

Source code in seedboxsync/cli/commands/cmd_user.py

@cli.command("edit", help="Edit an existing user.")  # type: ignore[untyped-decorator]
@click.option("--id", type=int, required=True, help="ID of the user to edit.")
@click.option("-u", "--username", help="New username for the account.")
@click.option("-e", "--email", help="New email address.")
@click.option("-p", "--password", help="New password for the account.")
@pass_context
def edit(ctx: Context, id: int, username: str | None, email: str | None, password: str | None) -> None:  # noqa: A002
    """
    Update details for an existing user account.

    Fetches the user by ID and updates only the provided fields. Password
    updates are automatically hashed before saving.

    Args:
        ctx (Context): The CLI application context.
        id (int): Database identifier of the user to edit.
        username (str | None): New username to set.
        email (str | None): New email address to set.
        password (str | None): New plain-text password to hash and update.
    """
    try:
        user = User.get_by_id(id)
    except User.DoesNotExist:  # pyright: ignore [reportAttributeAccessIssue]
        click.secho(f"Error: User with ID {id} does not exist.", fg="red", err=True)
        return

    # Check unique constraint collisions if username or email is being updated
    if username and username != user.username:
        if User.get_or_none(User.username == username):
            click.secho(f"Error: Username '{username}' is already taken.", fg="red", err=True)
            return
        user.username = username

    if email and email != user.email:
        if User.get_or_none(User.email == email):
            click.secho(f"Error: Email '{email}' is already taken.", fg="red", err=True)
            return
        user.email = email

    if password:
        user.password = generate_password_hash(password)

    try:
        user.save()
        click.secho(f"User '{user.username}' (ID: {user.id}) updated successfully.", fg="green")
    except Exception as e:
        click.secho(f"Error: Failed to update user {id}: {e}", fg="red", err=True)

list_user

list_user(ctx: Context, number: int, search: str) -> None

List registered users from the database.

Queries user accounts with optional text filtering on usernames and renders a formatted table containing user metadata.

Parameters:

ctx
(Context) –

The CLI application context.
number
(int) –

Maximum number of users to display. Defaults to 10.
search
(str) –

Optional search term to filter users by username.

Source code in seedboxsync/cli/commands/cmd_user.py

@cli.command("list", help="List users.")  # type: ignore[untyped-decorator]
@click.option("-n", "--number", type=int, default=10, help="Number of torrents to display.")
@click.option("-s", "--search", help="Term to search.")
@pass_context
def list_user(ctx: Context, number: int, search: str) -> None:
    """
    List registered users from the database.

    Queries user accounts with optional text filtering on usernames and renders
    a formatted table containing user metadata.

    Args:
        ctx (Context): The CLI application context.
        number (int): Maximum number of users to display. Defaults to 10.
        search (str): Optional search term to filter users by username.
    """
    # Build "where" expression
    conditions = []
    if search:
        conditions.append(User.username.contains(search))

    # DB query
    query = (
        User.select(
            User.id,
            User.username,
            User.email,
            User.origin,
            fn.short_datetime(User.created),
            fn.short_datetime(User.last_login),
        )
        .limit(number)
        .order_by(User.id.desc())
    )

    # if "where" expression
    if conditions:
        query = query.where(*conditions)
    data = query.dicts()

    click.echo(
        ctx.render(
            reversed(data),
            headers={"id": "Id", "username": "Username", "email": "Email", "origin": "Origin", "created": "Created", "last_login": "Last Login"},
        )
    )

context

Build a context used by Click.

Classes:

Context –

SeedboxSync Click context.
HeaderOptions –

Rich table column options.

Context


              flowchart TD
              seedboxsync.cli.context.Context[Context]

              

              click seedboxsync.cli.context.Context href "" "seedboxsync.cli.context.Context"

SeedboxSync Click context.

Used by:

seedboxsync cli commands
- cmd_health cli
- cmd_search
  - cli
  - downloaded
  - progress
  - uploaded
- cmd_stats
  - by_month
  - by_year
  - cli
  - total
- cmd_sync
  - blackhole
  - seedbox
- cmd_task
- cmd_user
  - add
  - cli
  - delete
  - edit
  - list_user

Methods:

render –

Render tabular data as a Rich table.

Attributes:

app (Flask) –

Return the current Flask application.

app `cached` `property`

app: Flask

Return the current Flask application.

Returns:

Flask ( Flask ) –

The current Flask application.

render

render(data: Iterable[Any], headers: Headers, title: str | None = None) -> str

Render tabular data as a Rich table.

Parameters:

data
(Iterable[Any]) –

Tabular data to render.
headers
(Headers) –

Column headers.
title
(str | None, default: None ) –

Table title.

Returns:

str ( str ) –

The formatted table..

Source code in seedboxsync/cli/context.py

def render(self, data: Iterable[Any], headers: Headers, title: str | None = None) -> str:
    """
    Render tabular data as a Rich table.

    Args:
        data: Tabular data to render.
        headers: Column headers.
        title: Table title.

    Returns:
        str: The formatted table..
    """
    console = Console()
    table = Table(title=title)

    # Set columns from headers
    for key, header in headers.items():
        if isinstance(header, str):
            column_title = header
            table.add_column(column_title)
            continue

        column_title = header.get("title", key)

        table.add_column(
            column_title,
            justify=header.get("justify", "left"),
            style=header.get("style"),
            no_wrap=header.get("no_wrap", False),
            overflow=header.get("overflow", "ellipsis"),
            width=header.get("width"),
            min_width=header.get("min_width"),
            max_width=header.get("max_width"),
            ratio=header.get("ratio"),
        )

    # Set rows from headers and data
    for row in data:
        table.add_row(*(str(row.get(column, "")) for column in headers))

    # Return as string
    with console.capture() as capture:
        console.print(table)
    return capture.get()

HeaderOptions


              flowchart TD
              seedboxsync.cli.context.HeaderOptions[HeaderOptions]

              

              click seedboxsync.cli.context.HeaderOptions href "" "seedboxsync.cli.context.HeaderOptions"

Rich table column options.

core

SeedboxSync Core package.

Modules:

config –

A module to manage SeedboxSync configuration from Database or environment variables.
database –

SeedboxSync database package.
exception –

Custom exception classes for SeedboxSync.
flask –

Flask core initer mobule.
logger –

Setup logger for Flask.
ping –

Ping package with all ping features.
sync –

SeedboxSync sync package.
taskmanager –

SeedboxSync taskmanager using Huey package.
utils –

A collection of utility functions for SeedboxSync.

Classes:

Config –

Config.
Database –

Database manager for SeedboxSync using Peewee ORM.
Flask –

Flask application with SeedboxSync-specific configuration helpers.

Config

Config(app: Flask, test_config: dict[str, str] | None = None)

Config.

Parameters:

app
(Flask) –

The Flask application to configure.
test_config
(dict[str, str] | None, default: None ) –

Configuration for testing.

Methods:

reload_config –

Reload application configuration from the database.

Source code in seedboxsync/core/config.py

def __init__(self, app: Flask, test_config: dict[str, str] | None = None) -> None:
    """
    Initialize a new Config instance.

    Args:
        app (Flask): The Flask application to configure.
        test_config (dict[str, str] | None): Configuration for testing.
    """
    self.app = app
    self.app.config.from_prefixed_env()  # Set from env prefixed by 'FLASK_'

    # Load config from database
    db_config = Config._load_config_from_database(self.app)
    self.app.config.from_mapping(db_config)

    self._check_config()  # Do all checks

    self.app.config.setdefault("CACHE_TYPE", "SimpleCache")  # Init Flask Cache
    self.app.config.setdefault("SWAGGER_UI_DOC_EXPANSION", "list")  # Expense swager namespaces
    self.app.config.setdefault("PROPAGATE_EXCEPTIONS", False)
    self.app.config.setdefault("LOGIN_DISABLED", self.app.config.get(Config.CONFIG_NAMESPACE + "LOGIN_DISABLED", False))  # Disable login
    self.app.config.setdefault("WTF_CSRF_ENABLED", self.app.config.get(Config.CONFIG_NAMESPACE + "WTF_CSRF_ENABLED", True))  # Disable CSRF

reload_config `staticmethod`

reload_config(app: Flask) -> dict[str, str]

Reload application configuration from the database.

Source code in seedboxsync/core/config.py

@staticmethod
def reload_config(app: Flask) -> dict[str, str]:
    """Reload application configuration from the database."""
    return Config._load_config_from_database(app)

Database

Database(app: Flask)

Database manager for SeedboxSync using Peewee ORM.

Handles database path resolution, connection binding, SQLite optimization pragmas, schema migrations, and custom SQLite functions.

Attributes:

DB_PATHS (ClassVar[list[Path]]) –

Candidate database paths checked in order of preference.
app (Flask) –

The Flask application instance bound to this database.
db (SqliteDatabase) –

The initialized Peewee SQLite database instance.

Parameters:

app
(Flask) –

The Flask application to bind to the database.

Source code in seedboxsync/core/database/database.py

def __init__(self, app: Flask) -> None:
    """
    Initialize a new Database instance.

    Args:
        app (Flask): The Flask application to bind to the database.
    """
    self.app = app
    self._load_database()
    self._register_functions()

Flask


              flowchart TD
              seedboxsync.core.Flask[Flask]

              

              click seedboxsync.core.Flask href "" "seedboxsync.core.Flask"

Flask application with SeedboxSync-specific configuration helpers.

Returned by:

seedboxsync
- create_app
- cli
  - Context app
  - context Context app

Attributes:

ping (AbstractPingClient) –

Return the configured ping client instance.
seedboxsync_config (dict[str, Any]) –

Return the SeedboxSync configuration namespace.
sync (AbstractSyncClient) –

Return the configured sync client instance.
task_manager (Manager) –

Return the task instance Manager.

ping `cached` `property`

ping: AbstractPingClient

Return the configured ping client instance.

Returns:

AbstractPingClient ( AbstractPingClient ) –

Configured ping client instance.

seedboxsync_config `property`

seedboxsync_config: dict[str, Any]

Return the SeedboxSync configuration namespace.

Returns:

dict[str, Any] –

The SeedboxSync configuration with the namespace prefix removed
dict[str, Any] –

and keys converted to lowercase.

sync `cached` `property`

sync: AbstractSyncClient

Return the configured sync client instance.

Returns:

AbstractSyncClient ( AbstractSyncClient ) –

Client instance.

task_manager `cached` `property`

task_manager: Manager

Return the task instance Manager.

Returns:

Manager –

The task manager instance.

config

A module to manage SeedboxSync configuration from Database or environment variables.

Classes:

Config –

Config.

Config

Config(app: Flask, test_config: dict[str, str] | None = None)

Config.

Parameters:

app
(Flask) –

The Flask application to configure.
test_config
(dict[str, str] | None, default: None ) –

Configuration for testing.

Methods:

reload_config –

Reload application configuration from the database.

Source code in seedboxsync/core/config.py

def __init__(self, app: Flask, test_config: dict[str, str] | None = None) -> None:
    """
    Initialize a new Config instance.

    Args:
        app (Flask): The Flask application to configure.
        test_config (dict[str, str] | None): Configuration for testing.
    """
    self.app = app
    self.app.config.from_prefixed_env()  # Set from env prefixed by 'FLASK_'

    # Load config from database
    db_config = Config._load_config_from_database(self.app)
    self.app.config.from_mapping(db_config)

    self._check_config()  # Do all checks

    self.app.config.setdefault("CACHE_TYPE", "SimpleCache")  # Init Flask Cache
    self.app.config.setdefault("SWAGGER_UI_DOC_EXPANSION", "list")  # Expense swager namespaces
    self.app.config.setdefault("PROPAGATE_EXCEPTIONS", False)
    self.app.config.setdefault("LOGIN_DISABLED", self.app.config.get(Config.CONFIG_NAMESPACE + "LOGIN_DISABLED", False))  # Disable login
    self.app.config.setdefault("WTF_CSRF_ENABLED", self.app.config.get(Config.CONFIG_NAMESPACE + "WTF_CSRF_ENABLED", True))  # Disable CSRF

reload_config `staticmethod`

reload_config(app: Flask) -> dict[str, str]

Reload application configuration from the database.

Source code in seedboxsync/core/config.py

@staticmethod
def reload_config(app: Flask) -> dict[str, str]:
    """Reload application configuration from the database."""
    return Config._load_config_from_database(app)

database

SeedboxSync database package.

Modules:

database –

Database module.
migration –

Database schema migration management.
models –

DAO package with all Peewee models.

Classes:

Database –

Database manager for SeedboxSync using Peewee ORM.
DatabaseMigration –

Manage database schema migrations.

Database

Database(app: Flask)

Database manager for SeedboxSync using Peewee ORM.

Handles database path resolution, connection binding, SQLite optimization pragmas, schema migrations, and custom SQLite functions.

Attributes:

DB_PATHS (ClassVar[list[Path]]) –

Candidate database paths checked in order of preference.
app (Flask) –

The Flask application instance bound to this database.
db (SqliteDatabase) –

The initialized Peewee SQLite database instance.

Parameters:

app
(Flask) –

The Flask application to bind to the database.

Source code in seedboxsync/core/database/database.py

def __init__(self, app: Flask) -> None:
    """
    Initialize a new Database instance.

    Args:
        app (Flask): The Flask application to bind to the database.
    """
    self.app = app
    self._load_database()
    self._register_functions()

DatabaseMigration

DatabaseMigration(app: Flask, db: SqliteDatabase)

Manage database schema migrations.

Parameters:

app
(Flask) –

Flask application instance.
db
(SqliteDatabase) –

Peewee SQLite database instance.

Methods:

upgrade –

Upgrade the database schema to the latest available migration.

Source code in seedboxsync/core/database/migration.py

def __init__(self, app: Flask, db: SqliteDatabase) -> None:
    """
    Initialize the database migration manager.

    Args:
        app: Flask application instance.
        db: Peewee SQLite database instance.
    """
    self.app = app
    self.db = db
    self.runner = Runner(self.db, directory=MIGRATION_PATH)

upgrade

upgrade() -> None

Upgrade the database schema to the latest available migration.

Existing databases using the legacy integer-based migration system are first converted to the Peewee migration history format.

Source code in seedboxsync/core/database/migration.py

def upgrade(self) -> None:
    """
    Upgrade the database schema to the latest available migration.

    Existing databases using the legacy integer-based migration system
    are first converted to the Peewee migration history format.
    """
    self._migrate_legacy_database()
    with self.app.app_context():
        self._run_migrations()
    self._set_last_migration()

database

Database module.

Classes:

Database –

Database manager for SeedboxSync using Peewee ORM.

Database

Database(app: Flask)

Database manager for SeedboxSync using Peewee ORM.

Handles database path resolution, connection binding, SQLite optimization pragmas, schema migrations, and custom SQLite functions.

Attributes:

DB_PATHS (ClassVar[list[Path]]) –

Candidate database paths checked in order of preference.
app (Flask) –

The Flask application instance bound to this database.
db (SqliteDatabase) –

The initialized Peewee SQLite database instance.

Parameters:

app
(Flask) –

The Flask application to bind to the database.

Source code in seedboxsync/core/database/database.py

def __init__(self, app: Flask) -> None:
    """
    Initialize a new Database instance.

    Args:
        app (Flask): The Flask application to bind to the database.
    """
    self.app = app
    self._load_database()
    self._register_functions()

migration

Database schema migration management.

Classes:

DatabaseMigration –

Manage database schema migrations.

DatabaseMigration

DatabaseMigration(app: Flask, db: SqliteDatabase)

Manage database schema migrations.

Parameters:

app
(Flask) –

Flask application instance.
db
(SqliteDatabase) –

Peewee SQLite database instance.

Methods:

upgrade –

Upgrade the database schema to the latest available migration.

Source code in seedboxsync/core/database/migration.py

def __init__(self, app: Flask, db: SqliteDatabase) -> None:
    """
    Initialize the database migration manager.

    Args:
        app: Flask application instance.
        db: Peewee SQLite database instance.
    """
    self.app = app
    self.db = db
    self.runner = Runner(self.db, directory=MIGRATION_PATH)

upgrade

upgrade() -> None

Upgrade the database schema to the latest available migration.

Existing databases using the legacy integer-based migration system are first converted to the Peewee migration history format.

Source code in seedboxsync/core/database/migration.py

def upgrade(self) -> None:
    """
    Upgrade the database schema to the latest available migration.

    Existing databases using the legacy integer-based migration system
    are first converted to the Peewee migration history format.
    """
    self._migrate_legacy_database()
    with self.app.app_context():
        self._run_migrations()
    self._set_last_migration()

models

DAO package with all Peewee models.

Modules:

apikey –

Peewee DAO model for ApiKey.
download –

Peewee DAO model for Download.
model –

Peewee model.
seedboxsync –

Peewee DAO model for SeedboxSync.
taskstatus –

Peewee DAO model for TaskStatus.
torrent –

Peewee DAO model for Torrent.
user –

Peewee DAO model for User.

Classes:

ApiKey –

Data Access Object (DAO) representing an API Key.
Download –

Data Access Object (DAO) representing a file download.
SeedboxSync –

Data Access Object (DAO) for application metadata and internal configuration.
SeedboxSyncModel –

Basemodel from which all other peewee models are derived.
TaskStatus –

Represents a taskstatus record in the system to prevent concurrent processes.
Torrent –

Data Access Object (DAO) representing a torrent.
User –

Data Access Object (DAO) representing a user.

ApiKey


              flowchart TD
              seedboxsync.core.database.models.ApiKey[ApiKey]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.ApiKey
                


              click seedboxsync.core.database.models.ApiKey href "" "seedboxsync.core.database.models.ApiKey"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing an API Key.

Stores API key metadata and hashed secret values associated with a user account for programmatic access.

Attributes:

id (int) –

Auto-incremented primary key.
user (User) –

Foreign key reference to the associated user account.
name (str) –

Human-readable label describing the purpose of the key.
key_hash (str) –

SHA-256 hash of the generated raw API key.
created (datetime) –

Timestamp when the API key was created.
last_used (datetime | None) –

Timestamp when the API key was last used.

Methods:

authenticate –

Authenticate an incoming API key string.
generate –

Generate a new API key for a user.

authenticate `classmethod`

authenticate(raw_key: str) -> User | None

Authenticate an incoming API key string.

Hashes the incoming raw key and verifies if a matching active key exists in the database. Updates last_used on success.

Parameters:

raw_key (str) –

Plain-text API key provided in the HTTP header/request.

Returns:

User | None –

User | None: The matching User instance, or None if authentication fails.

Source code in seedboxsync/core/database/models/apikey.py

@classmethod
def authenticate(cls, raw_key: str) -> User | None:
    """
    Authenticate an incoming API key string.

    Hashes the incoming raw key and verifies if a matching active key exists
    in the database. Updates `last_used` on success.

    Args:
        raw_key (str): Plain-text API key provided in the HTTP header/request.

    Returns:
        User | None: The matching User instance,
            or None if authentication fails.
    """
    if not raw_key or not raw_key.startswith(cls.KEY_PREFIX):
        return None

    key_hash = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
    api_key = cls.select(cls, User).join(User).where(cls.key_hash == key_hash).first()

    if api_key is None:
        return None

    # Update last_used timestamp
    api_key.last_used = datetime.datetime.now()
    api_key.save()

    return cast(User, api_key.user)

generate `classmethod`

generate(user: User, name: str) -> tuple[Self, str]

Generate a new API key for a user.

Computes a cryptographically secure random token, hashes it for storage, and returns both the database model instance and the raw token.

Parameters:

user (User) –

User model instance owning the new key.
name (str) –

Descriptive label for the key.

Returns:

tuple[Self, str] –

tuple[Self, str]: A tuple containing (ApiKey instance, raw_api_key_str). The raw key string must be displayed to the user immediately, as it cannot be recovered later.

Source code in seedboxsync/core/database/models/apikey.py

@classmethod
def generate(cls, user: User, name: str) -> tuple[Self, str]:
    """
    Generate a new API key for a user.

    Computes a cryptographically secure random token, hashes it for storage,
    and returns both the database model instance and the raw token.

    Args:
        user (User): User model instance owning the new key.
        name (str): Descriptive label for the key.

    Returns:
        tuple[Self, str]: A tuple containing (ApiKey instance, raw_api_key_str).
            The raw key string must be displayed to the user immediately,
            as it cannot be recovered later.
    """
    raw_key = f"{cls.KEY_PREFIX}{secrets.token_urlsafe(32)}"
    key_hash = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()

    api_key = cls.create(
        user=user,
        name=name,
        key_hash=key_hash,
    )

    return api_key, raw_key

Download


              flowchart TD
              seedboxsync.core.database.models.Download[Download]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.Download
                


              click seedboxsync.core.database.models.Download href "" "seedboxsync.core.database.models.Download"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing a file download.

This model stores information about a downloaded file, including its path, size on the seedbox and locally, as well as timestamps indicating when the download started and finished.

Used by:

seedboxsync core sync download_progress DownloadProgress

Methods:

get_stats_by_mime_type –

Get the total count and total size of downloaded files grouped by MIME type.
is_already_download –

Check if a file has already been downloaded.
set_mime –

Detect and update the MIME attributes of the downloaded file.

get_stats_by_mime_type `classmethod`

get_stats_by_mime_type(
    start_date: date | None = None, end_date: date | None = None
) -> list[dict[str, Any]]

Get the total count and total size of downloaded files grouped by MIME type.

Parameters:

start_date (date | None, default: None ) –

Optional start date filter.
end_date (date | None, default: None ) –

Optional end date filter.

Returns:

list[dict[str, Any]] –

list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.

Source code in seedboxsync/core/database/models/download.py

@classmethod
def get_stats_by_mime_type(cls, start_date: datetime.date | None = None, end_date: datetime.date | None = None) -> list[dict[str, Any]]:
    """
    Get the total count and total size of downloaded files grouped by MIME type.

    Args:
        start_date (datetime.date | None): Optional start date filter.
        end_date (datetime.date | None): Optional end date filter.

    Returns:
        list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.
    """
    # Build "where" expression
    conditions = []
    if start_date:
        conditions.append(cls.finished >= start_date)
    if end_date:
        conditions.append(cls.finished <= end_date)

    query = (
        cls.select(
            cls.mime_type,
            fn.COUNT(cls.id).alias("total"),
            fn.SUM(cls.local_size).alias("total_size"),
            fn.humanize(fn.SUM(cls.local_size)).alias("human_total_size"),
        )
        .group_by(cls.mime_type)
        .order_by(fn.SUM(cls.local_size).desc())
    )

    # if "where" expression
    if conditions:
        query = query.where(*conditions)
    data = query.dicts()

    return list(cast(Iterable[dict[str, Any]], data))

is_already_download `classmethod`

is_already_download(filepath: str) -> bool

Check if a file has already been downloaded.

Parameters:

filepath (str) –

Absolute or relative path to the file.

Returns:

bool ( bool ) –

True if the file was already downloaded (i.e. has a nonzero
bool –

finished timestamp), otherwise False.

Source code in seedboxsync/core/database/models/download.py

@classmethod
def is_already_download(cls, filepath: str) -> bool:
    """
    Check if a file has already been downloaded.

    Args:
        filepath (str): Absolute or relative path to the file.

    Returns:
        bool: True if the file was already downloaded (i.e. has a nonzero
        ``finished`` timestamp), otherwise False.
    """
    count = cls.select().where(cls.path == filepath, cls.finished > 0).count()
    return count != 0

set_mime

set_mime(save: bool = False) -> None

Detect and update the MIME attributes of the downloaded file.

Calls utils.get_mime_type_from_file to inspect the file path and header bytes, updating the mime_extension, mime_type, and mime_confidence attributes.

Parameters:

save (bool, default: False ) –

If True, saves the model instance to the database immediately after updating attributes. Defaults to False.

Source code in seedboxsync/core/database/models/download.py

def set_mime(self, save: bool = False) -> None:
    """
    Detect and update the MIME attributes of the downloaded file.

    Calls ``utils.get_mime_type_from_file`` to inspect the file path and header bytes,
    updating the ``mime_extension``, ``mime_type``, and ``mime_confidence`` attributes.

    Args:
        save (bool, optional): If True, saves the model instance to the database
            immediately after updating attributes. Defaults to False.
    """
    self.mime_type, self.mime_extension, self.mime_confidence = utils.get_mime_type_from_file(self.path)

    if save:
        self.save()

SeedboxSync


              flowchart TD
              seedboxsync.core.database.models.SeedboxSync[SeedboxSync]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.SeedboxSync
                


              click seedboxsync.core.database.models.SeedboxSync href "" "seedboxsync.core.database.models.SeedboxSync"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) for application metadata and internal configuration.

This table stores key-value pairs used for SeedboxSync's internal state management and configuration, such as taskstatus, versioning, or runtime parameters.

Attributes:

key (str) –

Unique identifier for the configuration entry.
value (str) –

Stored value associated with the key.

SeedboxSyncModel


              flowchart TD
              seedboxsync.core.database.models.SeedboxSyncModel[SeedboxSyncModel]

              

              click seedboxsync.core.database.models.SeedboxSyncModel href "" "seedboxsync.core.database.models.SeedboxSyncModel"

Basemodel from which all other peewee models are derived.

TaskStatus


              flowchart TD
              seedboxsync.core.database.models.TaskStatus[TaskStatus]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.TaskStatus
                


              click seedboxsync.core.database.models.TaskStatus href "" "seedboxsync.core.database.models.TaskStatus"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Represents a taskstatus record in the system to prevent concurrent processes.

Attributes:

key (str) –

Unique identifier for the task, e.g., 'sync_blackhole'.
running (bool) –

Indicates whether the task is currently running.
started (datetime) –

Timestamp when the task execution started.
finished (datetime) –

Timestamp when the task execution finished.

Torrent


              flowchart TD
              seedboxsync.core.database.models.Torrent[Torrent]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.Torrent
                


              click seedboxsync.core.database.models.Torrent href "" "seedboxsync.core.database.models.Torrent"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing a torrent.

This model stores metadata for a torrent file, including tracker info, source provenance, file counts, payload size, privacy status, and processing timestamp.

Attributes:

id (int) –

Auto-incremented primary key.
name (str) –

Name of the torrent.
announce (str) –

Tracker announce URL.
announcer (str) –

Tracker announce domain name.
source (str) –

Source or provenance tag of the torrent file.
total_files (int) –

Total number of files contained in the torrent.
total_size (int) –

Total size of all files in the torrent in bytes.
private (bool) –

Flag indicating if the torrent is marked as private.
sent (datetime) –

Timestamp indicating when the torrent was sent or created.

Methods:

get_stats_by_source –

Get the total count and total size of torrents grouped by source.
save –

Save the model instance, automatically updating the announcer domain.
set_from_file –

Populate the model from a torrent file.
update_announcer –

Extract and populate the announcer domain from the announce URL.

get_stats_by_source `classmethod`

get_stats_by_source(
    start_date: date | None = None, end_date: date | None = None
) -> list[dict[str, Any]]

Get the total count and total size of torrents grouped by source.

Parameters:

start_date (date | None, default: None ) –

Optional start date filter.
end_date (date | None, default: None ) –

Optional end date filter.

Returns:

list[dict[str, Any]] –

list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.

Source code in seedboxsync/core/database/models/torrent.py

@classmethod
def get_stats_by_source(cls, start_date: datetime.date | None = None, end_date: datetime.date | None = None) -> list[dict[str, Any]]:
    """
    Get the total count and total size of torrents grouped by source.

    Args:
        start_date (datetime.date | None): Optional start date filter.
        end_date (datetime.date | None): Optional end date filter.

    Returns:
        list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.
    """
    # Build "where" expression
    conditions = []
    if start_date:
        conditions.append(cls.sent >= start_date)
    if end_date:
        conditions.append(cls.sent <= end_date)

    query = (
        cls.select(
            cls.source,
            fn.COUNT(cls.id).alias("total"),
            fn.SUM(cls.total_size).alias("total_size"),
            fn.humanize(fn.SUM(cls.total_size)).alias("human_total_size"),
        )
        .group_by(cls.source)
        .order_by(fn.SUM(cls.total_size).desc())
    )

    # if "where" expression
    if conditions:
        query = query.where(*conditions)
    data = query.dicts()

    return list(cast(Iterable[dict[str, Any]], data))

save

save(force_insert: bool = False, only: Any | None = None) -> int

Save the model instance, automatically updating the announcer domain.

Source code in seedboxsync/core/database/models/torrent.py

def save(self, force_insert: bool = False, only: Any | None = None) -> int:
    """Save the model instance, automatically updating the announcer domain."""
    self.update_announcer()
    return super().save(force_insert=force_insert, only=only)

set_from_file

set_from_file(torrent_file: str | PathLike[str]) -> bool

Populate the model from a torrent file.

Parameters:

torrent_file (str | PathLike[str]) –

Path to the torrent file.

Returns:

bool –

True if the torrent is valid and minimal informations was successfully extracted,
bool –

otherwise False.

Source code in seedboxsync/core/database/models/torrent.py

def set_from_file(self, torrent_file: str | PathLike[str]) -> bool:
    """
    Populate the model from a torrent file.

    Args:
        torrent_file: Path to the torrent file.

    Returns:
        True if the torrent is valid and minimal informations was successfully extracted,
        otherwise False.
    """
    torrent_info = utils.get_torrent_infos(torrent_file)
    # Torrent is not valid
    if not isinstance(torrent_info, dict):
        return False

    # Get minimal information
    self.announce = torrent_info.get("announce") or None

    # Get extended informations
    info = torrent_info.get("info")
    if not isinstance(info, dict):
        return True

    self.source = info.get("source") or None
    self.private = info.get("private", False)

    self.total_files = None
    self.total_size = None
    files = info.get("files")
    if isinstance(files, list):
        # Multi-file torrent.
        valid_files = [file_info for file_info in files if isinstance(file_info, dict) and isinstance(file_info.get("length"), int)]
        self.total_files = len(valid_files)
        self.total_size = sum(file_info["length"] for file_info in valid_files)
    else:
        # Single-file torrent.
        length = info.get("length")
        self.total_files = 1 if isinstance(length, int) else None
        self.total_size = length if isinstance(length, int) else None

    return True

update_announcer

update_announcer() -> None

Extract and populate the announcer domain from the announce URL.

Parses announce URL and extracts the registrable domain (without subdomains).

Source code in seedboxsync/core/database/models/torrent.py

def update_announcer(self) -> None:
    """
    Extract and populate the announcer domain from the announce URL.

    Parses ``announce`` URL and extracts the registrable domain (without subdomains).
    """
    self.announcer = None  # reset on update
    if not self.announce:
        return

    hostname = urlparse(self.announce).hostname
    if not hostname:
        return

    # Use tldextract
    extracted = tldextract.extract(hostname)
    if not extracted.domain or not extracted.suffix:
        self.announcer = hostname
        return

    # Set announcer
    self.announcer = f"{extracted.domain}.{extracted.suffix}"

    # Source fallback
    if self.source == "" or self.source is None:
        self.source = self.announcer

User


              flowchart TD
              seedboxsync.core.database.models.User[User]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.User
                


              click seedboxsync.core.database.models.User href "" "seedboxsync.core.database.models.User"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing a user.

This model stores user account information, including authentication credentials, email address, account creation date, and last login timestamp.

Attributes:

id (int) –

Auto-incremented primary key.
username (str) –

Unique username of the user.
password (str) –

Hashed password of the user.
email (str) –

Unique email address of the user.
created (datetime) –

Timestamp when the user account was created.
last_login (datetime) –

Timestamp when the user last logged in.

Returned by:

seedboxsync core database models ApiKey authenticate

Used by:

seedboxsync core database models ApiKey generate

Methods:

authenticate –

Authenticate a user using their username or email address.
update_last_login –

Update the last login timestamp for the user.

authenticate `classmethod`

authenticate(login: str, password: str) -> Self | None

Authenticate a user using their username or email address.

Parameters:

login (str) –

Username or email address.
password (str) –

Plain-text password to verify.

Returns:

Self | None –

Self | None: The authenticated user, or None if authentication fails.

Source code in seedboxsync/core/database/models/user.py

@classmethod
def authenticate(cls, login: str, password: str) -> Self | None:
    """
    Authenticate a user using their username or email address.

    Args:
        login (str): Username or email address.
        password (str): Plain-text password to verify.

    Returns:
        Self | None: The authenticated user, or None if authentication fails.
    """
    user = cls.get_or_none((cls.username == login) | (cls.email == login))

    if user is None:
        return None

    if not check_password_hash(user.password, password):
        return None

    # Update last login timestamp
    user.update_last_login()

    return user

update_last_login

update_last_login() -> None

Update the last login timestamp for the user.

This method sets the last_login field to the current datetime and saves the change to the database.

Source code in seedboxsync/core/database/models/user.py

def update_last_login(self) -> None:
    """
    Update the last login timestamp for the user.

    This method sets the `last_login` field to the current datetime and
    saves the change to the database.
    """
    self.last_login = datetime.datetime.now()
    self.save()

apikey

Peewee DAO model for ApiKey.

Classes:

ApiKey –

Data Access Object (DAO) representing an API Key.

ApiKey


              flowchart TD
              seedboxsync.core.database.models.apikey.ApiKey[ApiKey]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.apikey.ApiKey
                


              click seedboxsync.core.database.models.apikey.ApiKey href "" "seedboxsync.core.database.models.apikey.ApiKey"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing an API Key.

Stores API key metadata and hashed secret values associated with a user account for programmatic access.

Attributes:

id (int) –

Auto-incremented primary key.
user (User) –

Foreign key reference to the associated user account.
name (str) –

Human-readable label describing the purpose of the key.
key_hash (str) –

SHA-256 hash of the generated raw API key.
created (datetime) –

Timestamp when the API key was created.
last_used (datetime | None) –

Timestamp when the API key was last used.

Methods:

authenticate –

Authenticate an incoming API key string.
generate –

Generate a new API key for a user.

authenticate classmethod

authenticate(raw_key: str) -> User | None

Authenticate an incoming API key string.

Hashes the incoming raw key and verifies if a matching active key exists in the database. Updates last_used on success.

Parameters:

raw_key (str) –

Plain-text API key provided in the HTTP header/request.

Returns:

User | None –

User | None: The matching User instance, or None if authentication fails.

Source code in seedboxsync/core/database/models/apikey.py

@classmethod
def authenticate(cls, raw_key: str) -> User | None:
    """
    Authenticate an incoming API key string.

    Hashes the incoming raw key and verifies if a matching active key exists
    in the database. Updates `last_used` on success.

    Args:
        raw_key (str): Plain-text API key provided in the HTTP header/request.

    Returns:
        User | None: The matching User instance,
            or None if authentication fails.
    """
    if not raw_key or not raw_key.startswith(cls.KEY_PREFIX):
        return None

    key_hash = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
    api_key = cls.select(cls, User).join(User).where(cls.key_hash == key_hash).first()

    if api_key is None:
        return None

    # Update last_used timestamp
    api_key.last_used = datetime.datetime.now()
    api_key.save()

    return cast(User, api_key.user)

generate classmethod

generate(user: User, name: str) -> tuple[Self, str]

Generate a new API key for a user.

Computes a cryptographically secure random token, hashes it for storage, and returns both the database model instance and the raw token.

Parameters:

user (User) –

User model instance owning the new key.
name (str) –

Descriptive label for the key.

Returns:

tuple[Self, str] –

tuple[Self, str]: A tuple containing (ApiKey instance, raw_api_key_str). The raw key string must be displayed to the user immediately, as it cannot be recovered later.

Source code in seedboxsync/core/database/models/apikey.py

@classmethod
def generate(cls, user: User, name: str) -> tuple[Self, str]:
    """
    Generate a new API key for a user.

    Computes a cryptographically secure random token, hashes it for storage,
    and returns both the database model instance and the raw token.

    Args:
        user (User): User model instance owning the new key.
        name (str): Descriptive label for the key.

    Returns:
        tuple[Self, str]: A tuple containing (ApiKey instance, raw_api_key_str).
            The raw key string must be displayed to the user immediately,
            as it cannot be recovered later.
    """
    raw_key = f"{cls.KEY_PREFIX}{secrets.token_urlsafe(32)}"
    key_hash = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()

    api_key = cls.create(
        user=user,
        name=name,
        key_hash=key_hash,
    )

    return api_key, raw_key

download

Peewee DAO model for Download.

Classes:

Download –

Data Access Object (DAO) representing a file download.

Download


              flowchart TD
              seedboxsync.core.database.models.download.Download[Download]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.download.Download
                


              click seedboxsync.core.database.models.download.Download href "" "seedboxsync.core.database.models.download.Download"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing a file download.

This model stores information about a downloaded file, including its path, size on the seedbox and locally, as well as timestamps indicating when the download started and finished.

Used by:

seedboxsync core sync download_progress DownloadProgress

Methods:

get_stats_by_mime_type –

Get the total count and total size of downloaded files grouped by MIME type.
is_already_download –

Check if a file has already been downloaded.
set_mime –

Detect and update the MIME attributes of the downloaded file.

get_stats_by_mime_type classmethod

get_stats_by_mime_type(
    start_date: date | None = None, end_date: date | None = None
) -> list[dict[str, Any]]

Get the total count and total size of downloaded files grouped by MIME type.

Parameters:

start_date (date | None, default: None ) –

Optional start date filter.
end_date (date | None, default: None ) –

Optional end date filter.

Returns:

list[dict[str, Any]] –

list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.

Source code in seedboxsync/core/database/models/download.py

@classmethod
def get_stats_by_mime_type(cls, start_date: datetime.date | None = None, end_date: datetime.date | None = None) -> list[dict[str, Any]]:
    """
    Get the total count and total size of downloaded files grouped by MIME type.

    Args:
        start_date (datetime.date | None): Optional start date filter.
        end_date (datetime.date | None): Optional end date filter.

    Returns:
        list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.
    """
    # Build "where" expression
    conditions = []
    if start_date:
        conditions.append(cls.finished >= start_date)
    if end_date:
        conditions.append(cls.finished <= end_date)

    query = (
        cls.select(
            cls.mime_type,
            fn.COUNT(cls.id).alias("total"),
            fn.SUM(cls.local_size).alias("total_size"),
            fn.humanize(fn.SUM(cls.local_size)).alias("human_total_size"),
        )
        .group_by(cls.mime_type)
        .order_by(fn.SUM(cls.local_size).desc())
    )

    # if "where" expression
    if conditions:
        query = query.where(*conditions)
    data = query.dicts()

    return list(cast(Iterable[dict[str, Any]], data))

is_already_download classmethod

is_already_download(filepath: str) -> bool

Check if a file has already been downloaded.

Parameters:

filepath (str) –

Absolute or relative path to the file.

Returns:

bool ( bool ) –

True if the file was already downloaded (i.e. has a nonzero
bool –

finished timestamp), otherwise False.

Source code in seedboxsync/core/database/models/download.py

@classmethod
def is_already_download(cls, filepath: str) -> bool:
    """
    Check if a file has already been downloaded.

    Args:
        filepath (str): Absolute or relative path to the file.

    Returns:
        bool: True if the file was already downloaded (i.e. has a nonzero
        ``finished`` timestamp), otherwise False.
    """
    count = cls.select().where(cls.path == filepath, cls.finished > 0).count()
    return count != 0

set_mime

set_mime(save: bool = False) -> None

Detect and update the MIME attributes of the downloaded file.

Calls utils.get_mime_type_from_file to inspect the file path and header bytes, updating the mime_extension, mime_type, and mime_confidence attributes.

Parameters:

save (bool, default: False ) –

If True, saves the model instance to the database immediately after updating attributes. Defaults to False.

Source code in seedboxsync/core/database/models/download.py

def set_mime(self, save: bool = False) -> None:
    """
    Detect and update the MIME attributes of the downloaded file.

    Calls ``utils.get_mime_type_from_file`` to inspect the file path and header bytes,
    updating the ``mime_extension``, ``mime_type``, and ``mime_confidence`` attributes.

    Args:
        save (bool, optional): If True, saves the model instance to the database
            immediately after updating attributes. Defaults to False.
    """
    self.mime_type, self.mime_extension, self.mime_confidence = utils.get_mime_type_from_file(self.path)

    if save:
        self.save()

model

Peewee model.

Classes:

SeedboxSyncModel –

Basemodel from which all other peewee models are derived.

SeedboxSyncModel


              flowchart TD
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

              

              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Basemodel from which all other peewee models are derived.

seedboxsync

Peewee DAO model for SeedboxSync.

Classes:

SeedboxSync –

Data Access Object (DAO) for application metadata and internal configuration.

SeedboxSync


              flowchart TD
              seedboxsync.core.database.models.seedboxsync.SeedboxSync[SeedboxSync]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.seedboxsync.SeedboxSync
                


              click seedboxsync.core.database.models.seedboxsync.SeedboxSync href "" "seedboxsync.core.database.models.seedboxsync.SeedboxSync"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) for application metadata and internal configuration.

This table stores key-value pairs used for SeedboxSync's internal state management and configuration, such as taskstatus, versioning, or runtime parameters.

Attributes:

key (str) –

Unique identifier for the configuration entry.
value (str) –

Stored value associated with the key.

taskstatus

Peewee DAO model for TaskStatus.

Classes:

TaskStatus –

Represents a taskstatus record in the system to prevent concurrent processes.

TaskStatus


              flowchart TD
              seedboxsync.core.database.models.taskstatus.TaskStatus[TaskStatus]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.taskstatus.TaskStatus
                


              click seedboxsync.core.database.models.taskstatus.TaskStatus href "" "seedboxsync.core.database.models.taskstatus.TaskStatus"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Represents a taskstatus record in the system to prevent concurrent processes.

Attributes:

key (str) –

Unique identifier for the task, e.g., 'sync_blackhole'.
running (bool) –

Indicates whether the task is currently running.
started (datetime) –

Timestamp when the task execution started.
finished (datetime) –

Timestamp when the task execution finished.

torrent

Peewee DAO model for Torrent.

Classes:

Torrent –

Data Access Object (DAO) representing a torrent.

Torrent


              flowchart TD
              seedboxsync.core.database.models.torrent.Torrent[Torrent]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.torrent.Torrent
                


              click seedboxsync.core.database.models.torrent.Torrent href "" "seedboxsync.core.database.models.torrent.Torrent"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing a torrent.

This model stores metadata for a torrent file, including tracker info, source provenance, file counts, payload size, privacy status, and processing timestamp.

Attributes:

id (int) –

Auto-incremented primary key.
name (str) –

Name of the torrent.
announce (str) –

Tracker announce URL.
announcer (str) –

Tracker announce domain name.
source (str) –

Source or provenance tag of the torrent file.
total_files (int) –

Total number of files contained in the torrent.
total_size (int) –

Total size of all files in the torrent in bytes.
private (bool) –

Flag indicating if the torrent is marked as private.
sent (datetime) –

Timestamp indicating when the torrent was sent or created.

Methods:

get_stats_by_source –

Get the total count and total size of torrents grouped by source.
save –

Save the model instance, automatically updating the announcer domain.
set_from_file –

Populate the model from a torrent file.
update_announcer –

Extract and populate the announcer domain from the announce URL.

get_stats_by_source classmethod

get_stats_by_source(
    start_date: date | None = None, end_date: date | None = None
) -> list[dict[str, Any]]

Get the total count and total size of torrents grouped by source.

Parameters:

start_date (date | None, default: None ) –

Optional start date filter.
end_date (date | None, default: None ) –

Optional end date filter.

Returns:

list[dict[str, Any]] –

list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.

Source code in seedboxsync/core/database/models/torrent.py

@classmethod
def get_stats_by_source(cls, start_date: datetime.date | None = None, end_date: datetime.date | None = None) -> list[dict[str, Any]]:
    """
    Get the total count and total size of torrents grouped by source.

    Args:
        start_date (datetime.date | None): Optional start date filter.
        end_date (datetime.date | None): Optional end date filter.

    Returns:
        list[dict[str, Any]]: A list of dicts containing mime_type, total count, and total_size.
    """
    # Build "where" expression
    conditions = []
    if start_date:
        conditions.append(cls.sent >= start_date)
    if end_date:
        conditions.append(cls.sent <= end_date)

    query = (
        cls.select(
            cls.source,
            fn.COUNT(cls.id).alias("total"),
            fn.SUM(cls.total_size).alias("total_size"),
            fn.humanize(fn.SUM(cls.total_size)).alias("human_total_size"),
        )
        .group_by(cls.source)
        .order_by(fn.SUM(cls.total_size).desc())
    )

    # if "where" expression
    if conditions:
        query = query.where(*conditions)
    data = query.dicts()

    return list(cast(Iterable[dict[str, Any]], data))

save

save(force_insert: bool = False, only: Any | None = None) -> int

Save the model instance, automatically updating the announcer domain.

Source code in seedboxsync/core/database/models/torrent.py

def save(self, force_insert: bool = False, only: Any | None = None) -> int:
    """Save the model instance, automatically updating the announcer domain."""
    self.update_announcer()
    return super().save(force_insert=force_insert, only=only)

set_from_file

set_from_file(torrent_file: str | PathLike[str]) -> bool

Populate the model from a torrent file.

Parameters:

torrent_file (str | PathLike[str]) –

Path to the torrent file.

Returns:

bool –

True if the torrent is valid and minimal informations was successfully extracted,
bool –

otherwise False.

Source code in seedboxsync/core/database/models/torrent.py

def set_from_file(self, torrent_file: str | PathLike[str]) -> bool:
    """
    Populate the model from a torrent file.

    Args:
        torrent_file: Path to the torrent file.

    Returns:
        True if the torrent is valid and minimal informations was successfully extracted,
        otherwise False.
    """
    torrent_info = utils.get_torrent_infos(torrent_file)
    # Torrent is not valid
    if not isinstance(torrent_info, dict):
        return False

    # Get minimal information
    self.announce = torrent_info.get("announce") or None

    # Get extended informations
    info = torrent_info.get("info")
    if not isinstance(info, dict):
        return True

    self.source = info.get("source") or None
    self.private = info.get("private", False)

    self.total_files = None
    self.total_size = None
    files = info.get("files")
    if isinstance(files, list):
        # Multi-file torrent.
        valid_files = [file_info for file_info in files if isinstance(file_info, dict) and isinstance(file_info.get("length"), int)]
        self.total_files = len(valid_files)
        self.total_size = sum(file_info["length"] for file_info in valid_files)
    else:
        # Single-file torrent.
        length = info.get("length")
        self.total_files = 1 if isinstance(length, int) else None
        self.total_size = length if isinstance(length, int) else None

    return True

update_announcer

update_announcer() -> None

Extract and populate the announcer domain from the announce URL.

Parses announce URL and extracts the registrable domain (without subdomains).

Source code in seedboxsync/core/database/models/torrent.py

def update_announcer(self) -> None:
    """
    Extract and populate the announcer domain from the announce URL.

    Parses ``announce`` URL and extracts the registrable domain (without subdomains).
    """
    self.announcer = None  # reset on update
    if not self.announce:
        return

    hostname = urlparse(self.announce).hostname
    if not hostname:
        return

    # Use tldextract
    extracted = tldextract.extract(hostname)
    if not extracted.domain or not extracted.suffix:
        self.announcer = hostname
        return

    # Set announcer
    self.announcer = f"{extracted.domain}.{extracted.suffix}"

    # Source fallback
    if self.source == "" or self.source is None:
        self.source = self.announcer

user

Peewee DAO model for User.

Classes:

User –

Data Access Object (DAO) representing a user.

User


              flowchart TD
              seedboxsync.core.database.models.user.User[User]
              seedboxsync.core.database.models.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.database.models.model.SeedboxSyncModel --> seedboxsync.core.database.models.user.User
                


              click seedboxsync.core.database.models.user.User href "" "seedboxsync.core.database.models.user.User"
              click seedboxsync.core.database.models.model.SeedboxSyncModel href "" "seedboxsync.core.database.models.model.SeedboxSyncModel"

Data Access Object (DAO) representing a user.

This model stores user account information, including authentication credentials, email address, account creation date, and last login timestamp.

Attributes:

id (int) –

Auto-incremented primary key.
username (str) –

Unique username of the user.
password (str) –

Hashed password of the user.
email (str) –

Unique email address of the user.
created (datetime) –

Timestamp when the user account was created.
last_login (datetime) –

Timestamp when the user last logged in.

Returned by:

seedboxsync core database models ApiKey authenticate

Used by:

seedboxsync core database models ApiKey generate

Methods:

authenticate –

Authenticate a user using their username or email address.
update_last_login –

Update the last login timestamp for the user.

authenticate classmethod

authenticate(login: str, password: str) -> Self | None

Authenticate a user using their username or email address.

Parameters:

login (str) –

Username or email address.
password (str) –

Plain-text password to verify.

Returns:

Self | None –

Self | None: The authenticated user, or None if authentication fails.

Source code in seedboxsync/core/database/models/user.py

@classmethod
def authenticate(cls, login: str, password: str) -> Self | None:
    """
    Authenticate a user using their username or email address.

    Args:
        login (str): Username or email address.
        password (str): Plain-text password to verify.

    Returns:
        Self | None: The authenticated user, or None if authentication fails.
    """
    user = cls.get_or_none((cls.username == login) | (cls.email == login))

    if user is None:
        return None

    if not check_password_hash(user.password, password):
        return None

    # Update last login timestamp
    user.update_last_login()

    return user

update_last_login

update_last_login() -> None

Update the last login timestamp for the user.

This method sets the last_login field to the current datetime and saves the change to the database.

Source code in seedboxsync/core/database/models/user.py

def update_last_login(self) -> None:
    """
    Update the last login timestamp for the user.

    This method sets the `last_login` field to the current datetime and
    saves the change to the database.
    """
    self.last_login = datetime.datetime.now()
    self.save()

exception

Custom exception classes for SeedboxSync.

This module defines the base error hierarchy used throughout the SeedboxSync application. All exceptions inherit from SeedboxSyncError, which handles logging and process termination in case of fatal errors.

Classes:

PingServiceError –

Exception raised when an unsupported or misconfigured ping service
SeedboxSyncConfigurationError –

Exception raised for configuration-related errors.
SeedboxSyncError –

Base exception class for all SeedboxSync errors.
SeedboxsyncConnectionError –

Exception raised when the connection to the remote seedbox fails.
SyncProtocoleError –

Exception raised when an unsupported or misconfigured synchronization protocol

PingServiceError

PingServiceError(msg: str)


              flowchart TD
              seedboxsync.core.exception.PingServiceError[PingServiceError]
              seedboxsync.core.exception.SeedboxSyncError[SeedboxSyncError]

                              seedboxsync.core.exception.SeedboxSyncError --> seedboxsync.core.exception.PingServiceError
                


              click seedboxsync.core.exception.PingServiceError href "" "seedboxsync.core.exception.PingServiceError"
              click seedboxsync.core.exception.SeedboxSyncError href "" "seedboxsync.core.exception.SeedboxSyncError"

Exception raised when an unsupported or misconfigured ping service is specified.

Source code in seedboxsync/core/exception.py

def __init__(self, msg: str) -> None:
    """SeedboxSyncError init."""
    logger.exception(msg)  # noqa LOG004
    sys.exit(msg)

SeedboxSyncConfigurationError

SeedboxSyncConfigurationError(msg: str)


              flowchart TD
              seedboxsync.core.exception.SeedboxSyncConfigurationError[SeedboxSyncConfigurationError]
              seedboxsync.core.exception.SeedboxSyncError[SeedboxSyncError]

                              seedboxsync.core.exception.SeedboxSyncError --> seedboxsync.core.exception.SeedboxSyncConfigurationError
                


              click seedboxsync.core.exception.SeedboxSyncConfigurationError href "" "seedboxsync.core.exception.SeedboxSyncConfigurationError"
              click seedboxsync.core.exception.SeedboxSyncError href "" "seedboxsync.core.exception.SeedboxSyncError"

Exception raised for configuration-related errors.

This error typically occurs when the configuration file contains invalid, missing, or inconsistent settings.

Source code in seedboxsync/core/exception.py

def __init__(self, msg: str) -> None:
    """SeedboxSyncError init."""
    logger.exception(msg)  # noqa LOG004
    sys.exit(msg)

SeedboxSyncError

SeedboxSyncError(msg: str)


              flowchart TD
              seedboxsync.core.exception.SeedboxSyncError[SeedboxSyncError]

              

              click seedboxsync.core.exception.SeedboxSyncError href "" "seedboxsync.core.exception.SeedboxSyncError"

Base exception class for all SeedboxSync errors.

When raised, this exception logs the error message and terminates the program immediately. It is intended for unrecoverable errors that prevent normal operation.

Parameters:

msg
(str) –

The error message to log and display before exiting.

Source code in seedboxsync/core/exception.py

def __init__(self, msg: str) -> None:
    """SeedboxSyncError init."""
    logger.exception(msg)  # noqa LOG004
    sys.exit(msg)

SeedboxsyncConnectionError

SeedboxsyncConnectionError(msg: str)


              flowchart TD
              seedboxsync.core.exception.SeedboxsyncConnectionError[SeedboxsyncConnectionError]
              seedboxsync.core.exception.SeedboxSyncError[SeedboxSyncError]

                              seedboxsync.core.exception.SeedboxSyncError --> seedboxsync.core.exception.SeedboxsyncConnectionError
                


              click seedboxsync.core.exception.SeedboxsyncConnectionError href "" "seedboxsync.core.exception.SeedboxsyncConnectionError"
              click seedboxsync.core.exception.SeedboxSyncError href "" "seedboxsync.core.exception.SeedboxSyncError"

Exception raised when the connection to the remote seedbox fails.

Source code in seedboxsync/core/exception.py

def __init__(self, msg: str) -> None:
    """SeedboxSyncError init."""
    logger.exception(msg)  # noqa LOG004
    sys.exit(msg)

SyncProtocoleError

SyncProtocoleError(msg: str)


              flowchart TD
              seedboxsync.core.exception.SyncProtocoleError[SyncProtocoleError]
              seedboxsync.core.exception.SeedboxSyncError[SeedboxSyncError]

                              seedboxsync.core.exception.SeedboxSyncError --> seedboxsync.core.exception.SyncProtocoleError
                


              click seedboxsync.core.exception.SyncProtocoleError href "" "seedboxsync.core.exception.SyncProtocoleError"
              click seedboxsync.core.exception.SeedboxSyncError href "" "seedboxsync.core.exception.SeedboxSyncError"

Exception raised when an unsupported or misconfigured synchronization protocol is specified.

Source code in seedboxsync/core/exception.py

def __init__(self, msg: str) -> None:
    """SeedboxSyncError init."""
    logger.exception(msg)  # noqa LOG004
    sys.exit(msg)

flask

Flask core initer mobule.

Classes:

SeedboxSyncFlask –

Flask application with SeedboxSync-specific configuration helpers.

SeedboxSyncFlask


              flowchart TD
              seedboxsync.core.flask.SeedboxSyncFlask[SeedboxSyncFlask]

              

              click seedboxsync.core.flask.SeedboxSyncFlask href "" "seedboxsync.core.flask.SeedboxSyncFlask"

Flask application with SeedboxSync-specific configuration helpers.

Returned by:

seedboxsync
- create_app
- cli
  - Context app
  - context Context app

Attributes:

ping (AbstractPingClient) –

Return the configured ping client instance.
seedboxsync_config (dict[str, Any]) –

Return the SeedboxSync configuration namespace.
sync (AbstractSyncClient) –

Return the configured sync client instance.
task_manager (Manager) –

Return the task instance Manager.

ping `cached` `property`

ping: AbstractPingClient

Return the configured ping client instance.

Returns:

AbstractPingClient ( AbstractPingClient ) –

Configured ping client instance.

seedboxsync_config `property`

seedboxsync_config: dict[str, Any]

Return the SeedboxSync configuration namespace.

Returns:

dict[str, Any] –

The SeedboxSync configuration with the namespace prefix removed
dict[str, Any] –

and keys converted to lowercase.

sync `cached` `property`

sync: AbstractSyncClient

Return the configured sync client instance.

Returns:

AbstractSyncClient ( AbstractSyncClient ) –

Client instance.

task_manager `cached` `property`

task_manager: Manager

Return the task instance Manager.

Returns:

Manager –

The task manager instance.

logger

Setup logger for Flask.

Classes:

ColorFormatter –

Format log records with colors based on their level.

Functions:

configure_logger –

Configure the logger by applying the custom formatter to all existing handlers.

ColorFormatter


              flowchart TD
              seedboxsync.core.logger.ColorFormatter[ColorFormatter]

              

              click seedboxsync.core.logger.ColorFormatter href "" "seedboxsync.core.logger.ColorFormatter"

Format log records with colors based on their level.

Methods:

format –

Format a log record and apply a color based on its log level.

format

format(record: LogRecord) -> str

Format a log record and apply a color based on its log level.

Parameters:

record
(LogRecord) –

The log record to format.

Returns:

str ( str ) –

The formatted and colorized log message.

Source code in seedboxsync/core/logger.py

def format(self, record: logging.LogRecord) -> str:
    """
    Format a log record and apply a color based on its log level.

    Args:
        record: The log record to format.

    Returns:
        str: The formatted and colorized log message.
    """
    message = super().format(record)
    color = self.COLORS.get(record.levelno)

    if color is None:
        return message

    return click.style(message, fg=color)

configure_logger

configure_logger(logger: Logger) -> None

Configure the logger by applying the custom formatter to all existing handlers.

Parameters:

logger
(Logger) –

The logger to configure.

Source code in seedboxsync/core/logger.py

def configure_logger(logger: logging.Logger) -> None:
    """
    Configure the logger by applying the custom formatter to all existing handlers.

    Args:
        logger: The logger to configure.
    """
    formatter = ColorFormatter(
        fmt="%(asctime)s [%(levelname)-8s] [%(module)s] %(message)s",
        datefmt="%H:%M:%S",
    )

    for handler in logger.handlers:
        handler.setFormatter(formatter)

ping

Ping package with all ping features.

Modules:

abstract_ping_client –

Abstract transport client using paramiko-like interface.
client –

Package with all ping clients.

Classes:

AbstractPingClient –

Abstract base class for transport clients.

AbstractPingClient

AbstractPingClient()

Abstract base class for transport clients.

All concrete clients must implement methods to manage file transfers and remote file operations.

Returned by:

seedboxsync core
- Flask ping
- flask SeedboxSyncFlask ping

Methods:

start –

Send a start ping for a given subcommand.
success –

Send a success ping for a given subcommand.

Source code in seedboxsync/core/ping/abstract_ping_client.py

@abstractmethod
def __init__(self) -> None:
    """Init method."""

start `abstractmethod`

start(sub_command: str) -> None

Send a start ping for a given subcommand.

Parameters:

sub_command
(str) –

The SeedboxSync subcommand to ping.

Source code in seedboxsync/core/ping/abstract_ping_client.py

@abstractmethod
def start(self, sub_command: str) -> None:
    """
    Send a start ping  for a given subcommand.

    Args:
        sub_command (str): The SeedboxSync subcommand to ping.
    """

success `abstractmethod`

success(sub_command: str) -> None

Send a success ping for a given subcommand.

Parameters:

sub_command
(str) –

The SeedboxSync subcommand to ping.

Source code in seedboxsync/core/ping/abstract_ping_client.py

@abstractmethod
def success(self, sub_command: str) -> None:
    """
    Send a success ping for a given subcommand.

    Args:
        sub_command (str): The SeedboxSync subcommand to ping.
    """

abstract_ping_client

Abstract transport client using paramiko-like interface.

This class defines the interface that all transport clients must implement, providing methods for file operations and session management on a remote server.

Classes:

AbstractPingClient –

Abstract base class for transport clients.

AbstractPingClient

AbstractPingClient()

Abstract base class for transport clients.

All concrete clients must implement methods to manage file transfers and remote file operations.

Returned by:

seedboxsync core
- Flask ping
- flask SeedboxSyncFlask ping

Methods:

start –

Send a start ping for a given subcommand.
success –

Send a success ping for a given subcommand.

Source code in seedboxsync/core/ping/abstract_ping_client.py

@abstractmethod
def __init__(self) -> None:
    """Init method."""

start `abstractmethod`

start(sub_command: str) -> None

Send a start ping for a given subcommand.

Parameters:

sub_command (str) –

The SeedboxSync subcommand to ping.

Source code in seedboxsync/core/ping/abstract_ping_client.py

@abstractmethod
def start(self, sub_command: str) -> None:
    """
    Send a start ping  for a given subcommand.

    Args:
        sub_command (str): The SeedboxSync subcommand to ping.
    """

success `abstractmethod`

success(sub_command: str) -> None

Send a success ping for a given subcommand.

Parameters:

sub_command (str) –

The SeedboxSync subcommand to ping.

Source code in seedboxsync/core/ping/abstract_ping_client.py

@abstractmethod
def success(self, sub_command: str) -> None:
    """
    Send a success ping for a given subcommand.

    Args:
        sub_command (str): The SeedboxSync subcommand to ping.
    """

client

Package with all ping clients.

Modules:

healthchecks –

Healthchecks management for SeedboxSync.

healthchecks

Healthchecks management for SeedboxSync.

Classes:

Healthchecks –

Class to manage Healthchecks's pings.

Healthchecks

Healthchecks()


              flowchart TD
              seedboxsync.core.ping.client.healthchecks.Healthchecks[Healthchecks]
              seedboxsync.core.ping.abstract_ping_client.AbstractPingClient[AbstractPingClient]

                              seedboxsync.core.ping.abstract_ping_client.AbstractPingClient --> seedboxsync.core.ping.client.healthchecks.Healthchecks
                


              click seedboxsync.core.ping.client.healthchecks.Healthchecks href "" "seedboxsync.core.ping.client.healthchecks.Healthchecks"
              click seedboxsync.core.ping.abstract_ping_client.AbstractPingClient href "" "seedboxsync.core.ping.abstract_ping_client.AbstractPingClient"

Class to manage Healthchecks's pings.

Methods:

start –

Send a start ping to Healthchecks for a given subcommand.
success –

Send a success ping to Healthchecks for a given subcommand.

Source code in seedboxsync/core/ping/client/healthchecks.py

def __init__(self) -> None:
    """Constructor for Healthchecks."""
    self.app = current_app

start

start(sub_command: str) -> None

Send a start ping to Healthchecks for a given subcommand.

Parameters:

sub_command (str) –

The SeedboxSync subcommand to ping.

Source code in seedboxsync/core/ping/client/healthchecks.py

def start(self, sub_command: str) -> None:
    """
    Send a start ping to Healthchecks for a given subcommand.

    Args:
        sub_command (str): The SeedboxSync subcommand to ping.
    """
    enabled = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_enabled")
    if enabled is False:
        self.app.logger.info(f'Healthchecks for "{sub_command}" disabled by configuration')
    else:
        base_url = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_ping_url", "")
        ping_url = f"{base_url.rstrip('/')}/start"
        self.app.logger.debug(f"Ping url: {ping_url}")

        try:
            urllib.request.urlopen(ping_url, timeout=10)
        except OSError:
            self.app.logger.exception("Healthchecks, ping failed")

success

success(sub_command: str) -> None

Send a success ping to Healthchecks for a given subcommand.

Parameters:

sub_command (str) –

The SeedboxSync subcommand to ping.

Source code in seedboxsync/core/ping/client/healthchecks.py

def success(self, sub_command: str) -> None:
    """
    Send a success ping to Healthchecks for a given subcommand.

    Args:
        sub_command (str): The SeedboxSync subcommand to ping.
    """
    enabled = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_enabled")
    if enabled is False:
        self.app.logger.info(f'Healthchecks for "{sub_command}" disabled by configuration')
    else:
        ping_url = self.app.seedboxsync_config.get("healthchecks_" + sub_command + "_ping_url", "")
        self.app.logger.debug(f"Ping url: {ping_url}")

        try:
            urllib.request.urlopen(ping_url, timeout=10)
        except OSError:
            self.app.logger.exception("Healthchecks, ping failed.")

sync

SeedboxSync sync package.

Modules:

abstract_sync_client –

Abstract transport client using paramiko-like interface.
client –

Package with all sync clients.
download_progress –

Callback for tracking and persisting download progress.
services –

Package with all sync servces.

Classes:

AbstractSyncClient –

Abstract base class for transport clients.

AbstractSyncClient

AbstractSyncClient()

Abstract base class for transport clients.

All concrete clients must implement methods to manage file transfers and remote file operations.

Returned by:

seedboxsync core
- Flask sync
- flask SeedboxSyncFlask sync

Methods:

chdir –

Change the current working directory on the remote session.
chmod –

Change permissions of a remote file.
close –

Close the transport session and release resources.
get –

Download a file from the remote server to the local host.
put –

Upload a local file to the remote server.
rename –

Rename a file or directory on the remote server.
stat –

Retrieve metadata about a file on the remote system.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def __init__(self) -> None:
    """Initialize the transport client."""

chdir `abstractmethod`

chdir(path: PathType) -> None

Change the current working directory on the remote session.

Parameters:

path
(PathType) –

New working directory path. Defaults to None.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def chdir(self, path: PathType) -> None:
    """
    Change the current working directory on the remote session.

    Args:
        path (PathType, optional): New working directory path. Defaults to None.
    """

chmod `abstractmethod`

chmod(path: PathType, mode: int) -> None

Change permissions of a remote file.

Permissions are unix-style, same as Python's os.chmod.

Parameters:

path
(PathType) –

Path to the file on the remote server.
mode
(int) –

New permissions to set.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def chmod(self, path: PathType, mode: int) -> None:
    """
    Change permissions of a remote file.

    Permissions are unix-style, same as Python's os.chmod.

    Args:
        path (PathType): Path to the file on the remote server.
        mode (int): New permissions to set.
    """

close `abstractmethod`

close() -> None

Close the transport session and release resources.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def close(self) -> None:
    """Close the transport session and release resources."""

get `abstractmethod`

get(
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None

Download a file from the remote server to the local host.

Parameters:

remote_path
(PathType) –

Path to the remote file to copy.
local_path
(PathType) –

Destination path on the local host.
progress_callback
(ProgressCallback | None, default: None ) –

Optional callback receiving bytes_transferred.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def get(
    self,
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None:
    """
    Download a file from the remote server to the local host.

    Args:
        remote_path (PathType): Path to the remote file to copy.
        local_path (PathType): Destination path on the local host.
        progress_callback (ProgressCallback | None): Optional callback receiving bytes_transferred.
    """

put `abstractmethod`

put(local_path: PathType, remote_path: PathType) -> Any

Upload a local file to the remote server.

Parameters:

local_path
(PathType) –

Path to the local file to copy.
remote_path
(PathType) –

Destination path on the server including filename. Specifying only a directory must raise an error.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def put(self, local_path: PathType, remote_path: PathType) -> Any:
    """
    Upload a local file to the remote server.

    Args:
        local_path (PathType): Path to the local file to copy.
        remote_path (PathType): Destination path on the server including filename.
                           Specifying only a directory must raise an error.
    """

rename `abstractmethod`

rename(old_path: PathType, new_path: PathType) -> None

Rename a file or directory on the remote server.

Parameters:

old_path
(stPathTyper) –

Existing path of the file or folder.
new_path
(PathType) –

New path or name for the file or folder.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def rename(self, old_path: PathType, new_path: PathType) -> None:
    """
    Rename a file or directory on the remote server.

    Args:
        old_path (stPathTyper): Existing path of the file or folder.
        new_path (PathType): New path or name for the file or folder.
    """

stat `abstractmethod`

stat(filepath: PathType) -> Any

Retrieve metadata about a file on the remote system.

Returns an object similar to Python's os.stat, with fewer fields. Supported fields: st_mode, st_size, st_uid, st_gid, st_atime, st_mtime.

Parameters:

filepath
(PathType) –

Path to the remote file.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def stat(self, filepath: PathType) -> Any:
    """
    Retrieve metadata about a file on the remote system.

    Returns an object similar to Python's os.stat, with fewer fields.
    Supported fields: st_mode, st_size, st_uid, st_gid, st_atime, st_mtime.

    Args:
        filepath (PathType): Path to the remote file.
    """

abstract_sync_client

Abstract transport client using paramiko-like interface.

This class defines the interface that all transport clients must implement, providing methods for file operations and session management on a remote server.

Classes:

AbstractSyncClient –

Abstract base class for transport clients.

AbstractSyncClient

AbstractSyncClient()

Abstract base class for transport clients.

All concrete clients must implement methods to manage file transfers and remote file operations.

Returned by:

seedboxsync core
- Flask sync
- flask SeedboxSyncFlask sync

Methods:

chdir –

Change the current working directory on the remote session.
chmod –

Change permissions of a remote file.
close –

Close the transport session and release resources.
get –

Download a file from the remote server to the local host.
put –

Upload a local file to the remote server.
rename –

Rename a file or directory on the remote server.
stat –

Retrieve metadata about a file on the remote system.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def __init__(self) -> None:
    """Initialize the transport client."""

chdir `abstractmethod`

chdir(path: PathType) -> None

Change the current working directory on the remote session.

Parameters:

path (PathType) –

New working directory path. Defaults to None.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def chdir(self, path: PathType) -> None:
    """
    Change the current working directory on the remote session.

    Args:
        path (PathType, optional): New working directory path. Defaults to None.
    """

chmod `abstractmethod`

chmod(path: PathType, mode: int) -> None

Change permissions of a remote file.

Permissions are unix-style, same as Python's os.chmod.

Parameters:

path (PathType) –

Path to the file on the remote server.
mode (int) –

New permissions to set.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def chmod(self, path: PathType, mode: int) -> None:
    """
    Change permissions of a remote file.

    Permissions are unix-style, same as Python's os.chmod.

    Args:
        path (PathType): Path to the file on the remote server.
        mode (int): New permissions to set.
    """

close `abstractmethod`

close() -> None

Close the transport session and release resources.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def close(self) -> None:
    """Close the transport session and release resources."""

get `abstractmethod`

get(
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None

Download a file from the remote server to the local host.

Parameters:

remote_path (PathType) –

Path to the remote file to copy.
local_path (PathType) –

Destination path on the local host.
progress_callback (ProgressCallback | None, default: None ) –

Optional callback receiving bytes_transferred.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def get(
    self,
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None:
    """
    Download a file from the remote server to the local host.

    Args:
        remote_path (PathType): Path to the remote file to copy.
        local_path (PathType): Destination path on the local host.
        progress_callback (ProgressCallback | None): Optional callback receiving bytes_transferred.
    """

put `abstractmethod`

put(local_path: PathType, remote_path: PathType) -> Any

Upload a local file to the remote server.

Parameters:

local_path (PathType) –

Path to the local file to copy.
remote_path (PathType) –

Destination path on the server including filename. Specifying only a directory must raise an error.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def put(self, local_path: PathType, remote_path: PathType) -> Any:
    """
    Upload a local file to the remote server.

    Args:
        local_path (PathType): Path to the local file to copy.
        remote_path (PathType): Destination path on the server including filename.
                           Specifying only a directory must raise an error.
    """

rename `abstractmethod`

rename(old_path: PathType, new_path: PathType) -> None

Rename a file or directory on the remote server.

Parameters:

old_path (stPathTyper) –

Existing path of the file or folder.
new_path (PathType) –

New path or name for the file or folder.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def rename(self, old_path: PathType, new_path: PathType) -> None:
    """
    Rename a file or directory on the remote server.

    Args:
        old_path (stPathTyper): Existing path of the file or folder.
        new_path (PathType): New path or name for the file or folder.
    """

stat `abstractmethod`

stat(filepath: PathType) -> Any

Retrieve metadata about a file on the remote system.

Returns an object similar to Python's os.stat, with fewer fields. Supported fields: st_mode, st_size, st_uid, st_gid, st_atime, st_mtime.

Parameters:

filepath (PathType) –

Path to the remote file.

Source code in seedboxsync/core/sync/abstract_sync_client.py

@abstractmethod
def stat(self, filepath: PathType) -> Any:
    """
    Retrieve metadata about a file on the remote system.

    Returns an object similar to Python's os.stat, with fewer fields.
    Supported fields: st_mode, st_size, st_uid, st_gid, st_atime, st_mtime.

    Args:
        filepath (PathType): Path to the remote file.
    """

client

Package with all sync clients.

Modules:

ftp –

Transport client using FTP protocol.
sftp –

Transport client using the SFTP protocol.

ftp

Transport client using FTP protocol.

Classes:

FtpClient –

FTP transport client using ftputil.
FtpSession –

FTP session factory for ftputil with explicit port and timeout support.

FtpClient

FtpClient()


              flowchart TD
              seedboxsync.core.sync.client.ftp.FtpClient[FtpClient]
              seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient[AbstractSyncClient]

                              seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient --> seedboxsync.core.sync.client.ftp.FtpClient
                


              click seedboxsync.core.sync.client.ftp.FtpClient href "" "seedboxsync.core.sync.client.ftp.FtpClient"
              click seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient href "" "seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient"

FTP transport client using ftputil.

Handles file transfers between NAS and Seedbox servers. Provides basic operations such as get, put, rename, chmod, and directory traversal.

Methods:

chdir –

Change the current working directory of the FTP session.
chmod –

Change the mode (permissions) of a remote file.
close –

Close the FTP client and underlying connection.
get –

Download a remote file from the FTP server.
put –

Upload a local file to the FTP server.
rename –

Rename a file or directory on the remote server.
stat –

Retrieve metadata for a remote file.
walk –

Walk through remote directories, yielding paths, folders, and files.

Source code in seedboxsync/core/sync/client/ftp.py

def __init__(self) -> None:
    """Initialize the FTP client with application connection parameters."""
    self.app = current_app

    # Get config
    config = self.app.seedboxsync_config

    self._host = config.get("seedbox_host", "")
    self._login = config.get("seedbox_login", "")
    self._password = config.get("seedbox_password", "")
    self._port = config.get("seedbox_port", "21")

    raw_timeout = config.get("seedbox_timeout", False)
    self._timeout = float(raw_timeout) if raw_timeout else None

    self._client = None

chdir

chdir(path: PathType | None = None) -> None

Change the current working directory of the FTP session.

Parameters:

path (Optional[PathType], default: None ) –

Target directory. If None, no change occurs.

Source code in seedboxsync/core/sync/client/ftp.py

def chdir(self, path: PathType | None = None) -> None:
    """
    Change the current working directory of the FTP session.

    Args:
        path (Optional[PathType]): Target directory. If None, no change occurs.
    """
    client = self._connect_before()
    if path is not None:
        client.chdir(fspath(path))

chmod

chmod(path: PathType, mode: int) -> None

Change the mode (permissions) of a remote file.

Parameters:

path (PathType) –

Path of the file.
mode (int) –

Unix-style permissions (like os.chmod).

Source code in seedboxsync/core/sync/client/ftp.py

def chmod(self, path: PathType, mode: int) -> None:
    """
    Change the mode (permissions) of a remote file.

    Args:
        path (PathType): Path of the file.
        mode (int): Unix-style permissions (like os.chmod).
    """
    client = self._connect_before()
    path = fspath(path)
    client.chmod(path, mode)

close

close() -> None

Close the FTP client and underlying connection.

Source code in seedboxsync/core/sync/client/ftp.py

def close(self) -> None:
    """Close the FTP client and underlying connection."""
    if self._client is not None:
        self.app.logger.debug("Close ftputil.FTPHost client")
        self._client.close()
        self._client = None

get

get(
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None

Download a remote file from the FTP server.

Parameters:

remote_path (PathType) –

Path of the remote file.
local_path (PathType) –

Destination path on the local host.
progress_callback (ProgressCallback | None, default: None ) –

Optional callback receiving bytes_transferred and total_bytes.

Source code in seedboxsync/core/sync/client/ftp.py

def get(
    self,
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None:
    """
    Download a remote file from the FTP server.

    Args:
        remote_path (PathType): Path of the remote file.
        local_path (PathType): Destination path on the local host.
        progress_callback (ProgressCallback | None): Optional callback receiving bytes_transferred and total_bytes.
    """
    client = self._connect_before()
    remote_path = fspath(remote_path)
    local_path = fspath(local_path)

    if progress_callback is None:
        client.download(remote_path, local_path)
        return

    total_size = client.stat(remote_path).st_size
    transferred = 0
    ftp_session = client._session

    with Path(local_path).open("wb") as local_file:

        def on_block(data: bytes) -> None:
            nonlocal transferred
            local_file.write(data)
            transferred += len(data)
            progress_callback(transferred, total_size)

        ftp_session.retrbinary(
            f"RETR {remote_path}",
            on_block,
        )

put

put(local_path: PathType, remote_path: PathType) -> None

Upload a local file to the FTP server.

Parameters:

local_path (PathType) –

Path to the local file.
remote_path (PathType) –

Destination path on the server (including filename).

Source code in seedboxsync/core/sync/client/ftp.py

def put(self, local_path: PathType, remote_path: PathType) -> None:
    """
    Upload a local file to the FTP server.

    Args:
        local_path (PathType): Path to the local file.
        remote_path (PathType): Destination path on the server (including filename).
    """
    client = self._connect_before()
    local_path = fspath(local_path)
    remote_path = fspath(remote_path)
    client.upload(local_path, remote_path)

rename

rename(old_path: PathType, new_path: PathType) -> None

Rename a file or directory on the remote server.

Parameters:

old_path (PathType) –

Existing path.
new_path (PathType) –

New path.

Source code in seedboxsync/core/sync/client/ftp.py

def rename(self, old_path: PathType, new_path: PathType) -> None:
    """
    Rename a file or directory on the remote server.

    Args:
        old_path (PathType): Existing path.
        new_path (PathType): New path.
    """
    client = self._connect_before()
    old_path = fspath(old_path)
    new_path = fspath(new_path)
    client.rename(old_path, new_path)

stat

stat(filepath: PathType) -> Any

Retrieve metadata for a remote file.

Parameters:

filepath (PathType) –

Remote file path.

Returns:

Any ( Any ) –

Object with attributes similar to Python's os.stat.

Source code in seedboxsync/core/sync/client/ftp.py

def stat(self, filepath: PathType) -> Any:
    """
    Retrieve metadata for a remote file.

    Args:
        filepath (PathType): Remote file path.

    Returns:
        Any: Object with attributes similar to Python's os.stat.
    """
    client = self._connect_before()
    filepath = fspath(filepath)

    return client.stat(filepath)

walk

walk(remote_path: PathType) -> Generator[tuple[str, list[str], list[str]]]

Walk through remote directories, yielding paths, folders, and files.

Parameters:

remote_path (PathType) –

Remote directory to traverse.

Yields:

Generator[tuple[str, list[str], list[str]]] –

tuple[str, list[str], list[str]]: (current_path, folders, files)

Source code in seedboxsync/core/sync/client/ftp.py

def walk(self, remote_path: PathType) -> Generator[tuple[str, list[str], list[str]]]:
    """
    Walk through remote directories, yielding paths, folders, and files.

    Args:
        remote_path (PathType): Remote directory to traverse.

    Yields:
        tuple[str, list[str], list[str]]: (current_path, folders, files)
    """
    client = self._connect_before()
    remote_path = fspath(remote_path)
    walk_path = remote_path if remote_path != "" else client.curdir
    for path, folders, files in client.walk(walk_path):
        if remote_path == "":
            path = self._remove_current_directory_prefix(client, path)
        yield path, folders, files

FtpSession

FtpSession(host: str, user: str, password: str, port: int = 21, timeout: float = -999)


              flowchart TD
              seedboxsync.core.sync.client.ftp.FtpSession[FtpSession]

              

              click seedboxsync.core.sync.client.ftp.FtpSession href "" "seedboxsync.core.sync.client.ftp.FtpSession"

FTP session factory for ftputil with explicit port and timeout support.

Parameters:

host (str) –

FTP server hostname.
user (str) –

FTP account username.
password (str) –

FTP account password.
port (int, default: 21 ) –

FTP server port.
timeout (float, default: -999 ) –

Connection timeout passed to ftplib.

Source code in seedboxsync/core/sync/client/ftp.py

def __init__(self, host: str, user: str, password: str, port: int = 21, timeout: float = -999) -> None:
    """
    Connect and authenticate an FTP session.

    Args:
        host (str): FTP server hostname.
        user (str): FTP account username.
        password (str): FTP account password.
        port (int): FTP server port.
        timeout (float): Connection timeout passed to ``ftplib``.
    """
    super().__init__()
    self.connect(host, port, timeout=timeout)
    self.login(user, password)

sftp

Transport client using the SFTP protocol.

Classes:

SftpClient –

SFTP transport client using Paramiko.

SftpClient

SftpClient()


              flowchart TD
              seedboxsync.core.sync.client.sftp.SftpClient[SftpClient]
              seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient[AbstractSyncClient]

                              seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient --> seedboxsync.core.sync.client.sftp.SftpClient
                


              click seedboxsync.core.sync.client.sftp.SftpClient href "" "seedboxsync.core.sync.client.sftp.SftpClient"
              click seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient href "" "seedboxsync.core.sync.abstract_sync_client.AbstractSyncClient"

SFTP transport client using Paramiko.

Handles file transfers between NAS and Seedbox servers. Provides basic operations such as get, put, rename, chmod, and directory traversal.

Methods:

chdir –

Change the current working directory of the SFTP session.
chmod –

Change the mode (permissions) of a remote file.
close –

Close the SFTP transport client and underlying connection.
get –

Download a remote file from the SFTP server.
put –

Upload a local file to the SFTP server.
rename –

Rename a file or directory on the remote server.
stat –

Retrieve metadata for a remote file.
walk –

Walk through remote directories, yielding paths, folders, and files.

Source code in seedboxsync/core/sync/client/sftp.py

def __init__(self) -> None:
    """Initialize the SFTP client with application connection parameters."""
    self.app = current_app

    # Get config
    config = self.app.seedboxsync_config

    self._host = config.get("seedbox_host", "")
    self._login = config.get("seedbox_login", "")
    self._password = config.get("seedbox_password", "")
    self._port = config.get("seedbox_port", "")

    raw_timeout = config.get("seedbox_timeout", False)
    self._timeout = float(raw_timeout) if raw_timeout else None

    self._max_concurrent_prefetch_requests = int(config.get("seedbox_max_concurrent_prefetch_requests", 128))
    self._transport = None

    self.app.logger.debug(f"Use sftp://{self._login}:****@{self._host}:{self._port}")

chdir

chdir(path: PathType | None = None) -> None

Change the current working directory of the SFTP session.

Parameters:

path (Optional[PathType], default: None ) –

Target directory. If None, no change occurs.

Source code in seedboxsync/core/sync/client/sftp.py

def chdir(self, path: PathType | None = None) -> None:
    """
    Change the current working directory of the SFTP session.

    Args:
        path (Optional[PathType]): Target directory. If None, no change occurs.
    """
    self._connect_before()
    path = fspath(path) if path is not None else None
    self._client.chdir(path)

chmod

chmod(path: PathType, mode: int) -> None

Change the mode (permissions) of a remote file.

Parameters:

path (PathType) –

Path of the file.
mode (int) –

Unix-style permissions (like os.chmod).

Source code in seedboxsync/core/sync/client/sftp.py

def chmod(self, path: PathType, mode: int) -> None:
    """
    Change the mode (permissions) of a remote file.

    Args:
        path (PathType): Path of the file.
        mode (int): Unix-style permissions (like os.chmod).
    """
    self._connect_before()
    path = fspath(path)
    self._client.chmod(path, mode)

close

close() -> None

Close the SFTP transport client and underlying connection.

Source code in seedboxsync/core/sync/client/sftp.py

def close(self) -> None:
    """Close the SFTP transport client and underlying connection."""
    if self._transport is not None:
        self.app.logger.debug("Close paramiko.Transport client")
        self._transport.close()

get

get(
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None

Download a remote file from the SFTP server.

Parameters:

remote_path (PathType) –

Path of the remote file.
local_path (PathType) –

Destination path on the local host.
progress_callback (ProgressCallback | None, default: None ) –

Optional callback receiving bytes_transferred.

Source code in seedboxsync/core/sync/client/sftp.py

def get(
    self,
    remote_path: PathType,
    local_path: PathType,
    progress_callback: ProgressCallback | None = None,
) -> None:
    """
    Download a remote file from the SFTP server.

    Args:
        remote_path (PathType): Path of the remote file.
        local_path (PathType): Destination path on the local host.
        progress_callback (ProgressCallback | None): Optional callback receiving bytes_transferred.
    """
    self._connect_before()
    remote_path = fspath(remote_path)
    local_path = fspath(local_path)

    self._client.get(
        remote_path,
        local_path,
        callback=progress_callback,
        max_concurrent_prefetch_requests=self._max_concurrent_prefetch_requests,
    )

put

put(local_path: PathType, remote_path: PathType) -> SFTPAttributes

Upload a local file to the SFTP server.

Parameters:

local_path (PathType) –

Path to the local file.
remote_path (PathType) –

Destination path on the server (including filename).

Returns:

SFTPAttributes ( SFTPAttributes ) –

Metadata of the uploaded file.

Source code in seedboxsync/core/sync/client/sftp.py

def put(self, local_path: PathType, remote_path: PathType) -> SFTPAttributes:
    """
    Upload a local file to the SFTP server.

    Args:
        local_path (PathType): Path to the local file.
        remote_path (PathType): Destination path on the server (including filename).

    Returns:
        SFTPAttributes: Metadata of the uploaded file.
    """
    self._connect_before()
    local_path = fspath(local_path)
    remote_path = fspath(remote_path)

    return self._client.put(local_path, remote_path)

rename

rename(old_path: PathType, new_path: PathType) -> None

Rename a file or directory on the remote server.

Parameters:

old_path (PathType) –

Existing path.
new_path (PathType) –

New path.

Source code in seedboxsync/core/sync/client/sftp.py

def rename(self, old_path: PathType, new_path: PathType) -> None:
    """
    Rename a file or directory on the remote server.

    Args:
        old_path (PathType): Existing path.
        new_path (PathType): New path.
    """
    self._connect_before()
    old_path = fspath(old_path)
    new_path = fspath(new_path)
    self._client.posix_rename(old_path, new_path)

stat

stat(filepath: PathType) -> SFTPAttributes

Retrieve metadata for a remote file.

Parameters:

filepath (PathType) –

Remote file path.

Returns:

SFTPAttributes ( SFTPAttributes ) –

Object with attributes similar to Python's os.stat: st_mode, st_size, st_uid, st_gid, st_atime, st_mtime.

Source code in seedboxsync/core/sync/client/sftp.py

def stat(self, filepath: PathType) -> SFTPAttributes:
    """
    Retrieve metadata for a remote file.

    Args:
        filepath (PathType): Remote file path.

    Returns:
        SFTPAttributes: Object with attributes similar to Python's os.stat:
            st_mode, st_size, st_uid, st_gid, st_atime, st_mtime.
    """
    self._connect_before()
    filepath = fspath(filepath)

    return self._client.stat(filepath)

walk

walk(remote_path: PathType) -> Generator[tuple[str, list[str], list[str]]]

Walk through remote directories, yielding paths, folders, and files.

Parameters:

remote_path (PathType) –

Remote directory to traverse.

Yields:

Generator[tuple[str, list[str], list[str]]] –

tuple[str, list[str], list[str]]: (current_path, folders, files)

Note

Simplified version of os.walk for SFTP. Efficient for large directories.

Source

https://gist.github.com/johnfink8/2190472

Source code in seedboxsync/core/sync/client/sftp.py

def walk(self, remote_path: PathType) -> Generator[tuple[str, list[str], list[str]]]:
    """
    Walk through remote directories, yielding paths, folders, and files.

    Args:
        remote_path (PathType): Remote directory to traverse.

    Yields:
        tuple[str, list[str], list[str]]: (current_path, folders, files)

    Note:
        Simplified version of os.walk for SFTP. Efficient for large directories.

    Source:
        https://gist.github.com/johnfink8/2190472
    """
    self._connect_before()
    remote_path = fspath(remote_path)
    path = remote_path
    files: list[str] = []
    folders: list[str] = []
    for f in self._client.listdir_attr(remote_path):
        if f.st_mode is not None and S_ISDIR(f.st_mode):
            folders.append(f.filename)
        else:
            files.append(f.filename)
    yield path, folders, files

    for folder in folders:
        new_path = str(Path(remote_path) / folder)
        yield from self.walk(new_path)

download_progress

Callback for tracking and persisting download progress.

Classes:

DownloadProgress –

Track download progress and update its database record.

DownloadProgress

DownloadProgress(download: Download)

Track download progress and update its database record.

Parameters:

download
(Download) –

The download record to update.

Methods:

__call__ –

Update the download progress and persist it when appropriate.

Source code in seedboxsync/core/sync/download_progress.py

def __init__(self, download: Download) -> None:
    """
    Initialize the download progress tracker.

    Args:
        download: The download record to update.
    """
    self.app = current_app
    self._download = download
    self._last_saved_size = 0

call

__call__(transferred: int, total: int) -> None

Update the download progress and persist it when appropriate.

Parameters:

transferred (int) –

Number of bytes transferred so far.
total (int) –

Total number of bytes to transfer.

Source code in seedboxsync/core/sync/download_progress.py

def __call__(self, transferred: int, total: int) -> None:
    """
    Update the download progress and persist it when appropriate.

    Args:
        transferred: Number of bytes transferred so far.
        total: Total number of bytes to transfer.
    """
    # Persist progress every 100 MiB and when the download completes.
    if transferred == total or transferred - self._last_saved_size >= 100 * 1024 * 1024:
        heartbeat()  # Call heartbeat
        self._download.local_size = transferred
        percent = (transferred / total) * 100 if total > 0 else 0
        self.app.logger.debug(f"Download progress: {transferred} / {total} ({percent:.2f}%)")
        self._download.save()
        self._last_saved_size = transferred

services

Package with all sync servces.

Modules:

blackhole –

SeedboxSync sync service for blackhole.
seedbox –

SeedboxSync sync service for seedbox.

blackhole

SeedboxSync sync service for blackhole.

Functions:

blackhole –

Perform the blackhole synchronization.

blackhole

blackhole(dry_run: bool, ping: bool) -> None

Perform the blackhole synchronization.

Uploads torrent files from the local watch folder to the seedbox. Optional dry-run, file permissions, database persistence, and error handling.

Parameters:

dry_run (bool) –

Whether to perform a dry run.
ping (bool) –

Whether to ping a service during execution.

Source code in seedboxsync/core/sync/services/blackhole.py

@track_taskstatus(LOCK_NAME)
def blackhole(dry_run: bool, ping: bool) -> None:
    """
    Perform the blackhole synchronization.

    Uploads torrent files from the local watch folder to the seedbox.
    Optional dry-run, file permissions, database persistence, and error handling.

    Args:
        dry_run (bool): Whether to perform a dry run.
        ping (bool): Whether to ping a service during execution.
    """
    if not current_app.seedboxsync_config.get("sync_blackhole_enabled"):
        current_app.logger.info("Blackhole synchronization task is disabled")
        return

    current_app.logger.debug(f'sync blackhole dry-run: "{dry_run}"')
    current_app.logger.debug(f'sync blackhole ping: "{ping}"')

    # Call ping.start() if enabled
    if ping:
        current_app.ping.start("sync_blackhole")

    # Gather all torrent files
    local_watch_path = current_app.seedboxsync_config.get("local_watch_path", "")
    current_app.logger.debug(f'Scanning for torrent files in "{local_watch_path}"')
    torrents = list(Path(local_watch_path).expanduser().resolve().glob("*.torrent"))

    if len(torrents) == 0:
        current_app.logger.info('No torrent files found in "{}"'.format(current_app.seedboxsync_config.get("local_watch_path")))
        # Call ping.success() if enabled
        if ping:
            current_app.ping.success("sync_blackhole")
        return

    for torrent_file in torrents:
        torrent_name = torrent_file.name

        # Dry-run mode
        if dry_run:
            current_app.logger.info(f'Dry-run: not uploading torrent "{torrent_name}"')
            continue

        tmp_path = current_app.seedboxsync_config.get("seedbox_tmp_path", "")
        watch_path = current_app.seedboxsync_config.get("seedbox_watch_path", "")

        current_app.logger.info(f'Upload torrent: "{torrent_name}"')
        current_app.logger.debug(f'Upload "{torrent_file}" to "{tmp_path}"')

        try:
            current_app.sync.chdir(None)  # type: ignore[arg-type]
            current_app.sync.put(torrent_file, Path(tmp_path) / torrent_name)

            # Apply chmod if configured
            chmod = current_app.seedboxsync_config.get("seedbox_chmod", False)
            if isinstance(chmod, str):
                current_app.logger.debug(f"Change permissions to {chmod}")
                current_app.sync.chmod(Path(tmp_path) / torrent_name, int(chmod, 8))

            # Move file from tmp to watch directory
            current_app.logger.debug(f'Move from "{tmp_path}" to "{watch_path}"')
            current_app.sync.rename(
                Path(tmp_path) / torrent_name,
                Path(watch_path) / torrent_name,
            )

            # Store torrent info in database
            torrent = Torrent.create(name=torrent_name)
            if torrent.set_from_file(torrent_file):
                torrent.save()

                # Remove local torrent file
                current_app.logger.debug(f'Remove local torrent "{torrent_file}"')
                Path(torrent_file).unlink()
            else:
                current_app.logger.warning(f'Rename local "{torrent_file}" to .torrent.fail')
                Path(torrent_file).rename(fspath(torrent_file) + ".fail")
        except SSHException as exc:
            current_app.logger.warning(f"SSH client exception > {exc!s}")

    # Call ping.success() if enabled
    if ping:
        current_app.ping.success("sync_blackhole")

seedbox

SeedboxSync sync service for seedbox.

Functions:

seedbox –

Perform synchronization from the seedbox.

seedbox

seedbox(dry_run: bool, ping: bool, only_store: bool) -> None

Perform synchronization from the seedbox.

Downloads files from the remote seedbox to the local machine, supports optional dry-run and only-store modes, applies exclusion patterns, and persists download information in the database.

Parameters:

dry_run (bool) –

Whether to list files without downloading or persisting them.
ping (bool) –

Whether to ping the configured monitoring service.
only_store (bool) –

Whether to record remote files without downloading them.

Source code in seedboxsync/core/sync/services/seedbox.py

@track_taskstatus(LOCK_NAME)
def seedbox(dry_run: bool, ping: bool, only_store: bool) -> None:
    """
    Perform synchronization from the seedbox.

    Downloads files from the remote seedbox to the local machine,
    supports optional dry-run and only-store modes, applies exclusion patterns,
    and persists download information in the database.

    Args:
        dry_run (bool): Whether to list files without downloading or persisting them.
        ping (bool): Whether to ping the configured monitoring service.
        only_store (bool): Whether to record remote files without downloading them.
    """
    if not current_app.seedboxsync_config.get("sync_seedbox_enabled"):
        current_app.logger.info("Seedbox synchronization task is disabled")
        return

    current_app.logger.debug(f'sync seedbox dry-run: "{dry_run}"')
    current_app.logger.debug(f'sync seeddbox only-store: "{only_store}"')
    current_app.logger.debug(f'sync seedbox ping: "{ping}"')

    # Call ping.start() if enabled
    if ping:
        current_app.ping.start("sync_seedbox")

    finished_path = current_app.seedboxsync_config.get("seedbox_finished_path", "")
    part_suffix = current_app.seedboxsync_config.get("seedbox_part_suffix")
    current_app.logger.debug(f'Scanning files in "{finished_path}"')

    # Walk through all files on the seedbox
    try:
        try:
            current_app.sync.chdir(None)  # type: ignore[arg-type]
            current_app.sync.chdir(finished_path)
        except FileNotFoundError as exc:
            current_app.logger.error(f"{exc!s}\nFailed to scan directory: {finished_path}")
            return

        for root, _, filenames in current_app.sync.walk(""):  # type: ignore[attr-defined]
            root = Path(root)

            for filename in filenames:
                filepath = root / filename

                if filepath.suffix == part_suffix:
                    current_app.logger.debug(f'Skipping part file "{filename}"')
                elif Download.is_already_download(filepath):
                    current_app.logger.debug(f'Skipping already downloaded file "{filename}"')
                elif __exclude_by_pattern(filepath):
                    current_app.logger.debug(f'Skipping excluded file "{filename}"')
                else:
                    if dry_run:
                        current_app.logger.info(f'Dry-run: not downloading "{filepath}"')
                        continue

                    __get_file(filepath, only_store)

    except (OSError, FileNotFoundError) as exc:
        current_app.logger.error(f'SeedboxSyncError > "{exc}"')

    # Call ping.success() if enabled
    if ping:
        current_app.ping.success("sync_seedbox")

taskmanager

SeedboxSync taskmanager using Huey package.

Modules:

manager –

Taskmanager Manager module.
task –

Package with all tasks.
track_taskstatus –

Track task status by decorator.
utils –

Task queue initialization and automatic task module registration.

Classes:

Manager –

Manage the Huey task queue integration with Flask.

Functions:

heartbeat –

Task manager heartbeat.
heartbeat_shutdown –

Task manager heartbeat.
heartbeat_startup –

Task manager heartbeat.

Manager

Manager(app: Flask | None = None)

Manage the Huey task queue integration with Flask.

The manager creates a Huey instance from the application configuration, stores it in the Flask extensions registry, and proxies Huey attributes through the manager instance.

Parameters:

app
(Flask | None, default: None ) –

Optional Flask application to initialize immediately.

Returned by:

seedboxsync core
- Flask task_manager
- flask SeedboxSyncFlask task_manager

Methods:

__getattr__ –

Proxy unknown attributes to the underlying Huey instance.
init_app –

Initialize the Huey manager for a Flask application.
init_huey –

Create and configure the Huey task queue instance.

Source code in seedboxsync/core/taskmanager/manager.py

def __init__(self, app: Flask | None = None) -> None:
    """
    Initialize the Huey manager.

    Args:
        app: Optional Flask application to initialize immediately.
    """
    self.app: Flask | None = None
    self.__instance: SqliteHuey | None = None

    if app is not None:
        self.init_app(app)

getattr

__getattr__(name: str) -> Any

Proxy unknown attributes to the underlying Huey instance.

Parameters:

name
(str) –

Name of the requested attribute.

Returns:

Any –

The attribute exposed by the underlying Huey instance.

Raises:

RuntimeError –

If the manager has not been initialized.
AttributeError –

If the Huey instance does not expose the requested attribute.

Source code in seedboxsync/core/taskmanager/manager.py

def __getattr__(self, name: str) -> Any:
    """
    Proxy unknown attributes to the underlying Huey instance.

    Args:
        name: Name of the requested attribute.

    Returns:
        The attribute exposed by the underlying Huey instance.

    Raises:
        RuntimeError: If the manager has not been initialized.
        AttributeError: If the Huey instance does not expose the requested
            attribute.
    """
    if self.__instance is None:
        raise RuntimeError("The Huey manager has not been initialized with a Flask application.")

    return getattr(self.__instance, name)

init_app

init_app(app: Flask) -> None

Initialize the Huey manager for a Flask application.

The created Huey instance is stored in the Flask extensions registry under the huey key.

Parameters:

app
(Flask) –

Flask application to initialize.

Source code in seedboxsync/core/taskmanager/manager.py

def init_app(self, app: Flask) -> None:
    """
    Initialize the Huey manager for a Flask application.

    The created Huey instance is stored in the Flask extensions registry
    under the ``huey`` key.

    Args:
        app: Flask application to initialize.
    """
    self.app = app

    huey_instance = self.init_huey()
    app.extensions["huey"] = huey_instance
    self.__instance = huey_instance

init_huey

init_huey() -> SqliteHuey

Create and configure the Huey task queue instance.

Returns:

SqliteHuey –

The configured SQLite-backed Huey instance.

Source code in seedboxsync/core/taskmanager/manager.py

def init_huey(self) -> SqliteHuey:
    """
    Create and configure the Huey task queue instance.

    Returns:
        The configured SQLite-backed Huey instance.
    """
    return SqliteHuey(
        self.HUEY_APP_NAME,
        filename=str(self._get_huey_database()),
    )

heartbeat

heartbeat() -> None

Task manager heartbeat.

Update the TaskStatus with key "heartbeat".

Source code in seedboxsync/core/taskmanager/track_taskstatus.py

def heartbeat() -> None:
    """
    Task manager heartbeat.

    Update the TaskStatus with key "heartbeat".
    """
    finished = datetime.now()
    TaskStatus.insert(
        key="heartbeat",
        finished=finished,
    ).on_conflict(
        conflict_target=[TaskStatus.key],
        update={
            TaskStatus.finished: finished,
        },
    ).execute()

heartbeat_shutdown

heartbeat_shutdown() -> None

Task manager heartbeat.

Update the TaskStatus with key "heartbeat".

Source code in seedboxsync/core/taskmanager/track_taskstatus.py

def heartbeat_shutdown() -> None:
    """
    Task manager heartbeat.

    Update the TaskStatus with key "heartbeat".
    """
    finished = datetime.now()
    TaskStatus.insert(
        key="heartbeat",
        running=False,
        finished=finished,
    ).on_conflict(
        conflict_target=[TaskStatus.key],
        update={
            TaskStatus.running: False,
            TaskStatus.finished: finished,
        },
    ).execute()

heartbeat_startup

heartbeat_startup() -> None

Task manager heartbeat.

Update the TaskStatus with key "heartbeat".

Source code in seedboxsync/core/taskmanager/track_taskstatus.py

def heartbeat_startup() -> None:
    """
    Task manager heartbeat.

    Update the TaskStatus with key "heartbeat".
    """
    started = datetime.now()
    TaskStatus.insert(
        key="heartbeat",
        running=True,
        started=started,
        finished=started,
    ).on_conflict(
        conflict_target=[TaskStatus.key],
        update={
            TaskStatus.running: True,
            TaskStatus.started: started,
            TaskStatus.finished: started,
        },
    ).execute()

manager

Taskmanager Manager module.

Classes:

Manager –

Manage the Huey task queue integration with Flask.

Manager

Manager(app: Flask | None = None)

Manage the Huey task queue integration with Flask.

The manager creates a Huey instance from the application configuration, stores it in the Flask extensions registry, and proxies Huey attributes through the manager instance.

Parameters:

app
(Flask | None, default: None ) –

Optional Flask application to initialize immediately.

Returned by:

seedboxsync core
- Flask task_manager
- flask SeedboxSyncFlask task_manager

Methods:

__getattr__ –

Proxy unknown attributes to the underlying Huey instance.
init_app –

Initialize the Huey manager for a Flask application.
init_huey –

Create and configure the Huey task queue instance.

Source code in seedboxsync/core/taskmanager/manager.py

def __init__(self, app: Flask | None = None) -> None:
    """
    Initialize the Huey manager.

    Args:
        app: Optional Flask application to initialize immediately.
    """
    self.app: Flask | None = None
    self.__instance: SqliteHuey | None = None

    if app is not None:
        self.init_app(app)

getattr

__getattr__(name: str) -> Any

Proxy unknown attributes to the underlying Huey instance.

Parameters:

name (str) –

Name of the requested attribute.

Returns:

Any –

The attribute exposed by the underlying Huey instance.

Raises:

RuntimeError –

If the manager has not been initialized.
AttributeError –

If the Huey instance does not expose the requested attribute.

Source code in seedboxsync/core/taskmanager/manager.py

def __getattr__(self, name: str) -> Any:
    """
    Proxy unknown attributes to the underlying Huey instance.

    Args:
        name: Name of the requested attribute.

    Returns:
        The attribute exposed by the underlying Huey instance.

    Raises:
        RuntimeError: If the manager has not been initialized.
        AttributeError: If the Huey instance does not expose the requested
            attribute.
    """
    if self.__instance is None:
        raise RuntimeError("The Huey manager has not been initialized with a Flask application.")

    return getattr(self.__instance, name)

init_app

init_app(app: Flask) -> None

Initialize the Huey manager for a Flask application.

The created Huey instance is stored in the Flask extensions registry under the huey key.

Parameters:

app (Flask) –

Flask application to initialize.

Source code in seedboxsync/core/taskmanager/manager.py

def init_app(self, app: Flask) -> None:
    """
    Initialize the Huey manager for a Flask application.

    The created Huey instance is stored in the Flask extensions registry
    under the ``huey`` key.

    Args:
        app: Flask application to initialize.
    """
    self.app = app

    huey_instance = self.init_huey()
    app.extensions["huey"] = huey_instance
    self.__instance = huey_instance

init_huey

init_huey() -> SqliteHuey

Create and configure the Huey task queue instance.

Returns:

SqliteHuey –

The configured SQLite-backed Huey instance.

Source code in seedboxsync/core/taskmanager/manager.py

def init_huey(self) -> SqliteHuey:
    """
    Create and configure the Huey task queue instance.

    Returns:
        The configured SQLite-backed Huey instance.
    """
    return SqliteHuey(
        self.HUEY_APP_NAME,
        filename=str(self._get_huey_database()),
    )

task

Package with all tasks.

Modules:

task_periodic_sync_blackhole –

Define a huey tasks for blackhole synchronization.
task_periodic_sync_seedbox –

Define a huey tasks for seedbox synchronization.
task_sync_blackhole –

Define a huey tasks for blackhole synchronization.
task_sync_seedbox –

Define a huey tasks for seedbox synchronization.

task_periodic_sync_blackhole

Define a huey tasks for blackhole synchronization.

Functions:

periodic_sync_blackhole –

Define a huey periodic task.

periodic_sync_blackhole

periodic_sync_blackhole() -> None

Define a huey periodic task.

Source code in seedboxsync/core/taskmanager/task/task_periodic_sync_blackhole.py

@task_manager.periodic_task(crontab(minute=minute), priority=PRIORITY)  # type: ignore[untyped-decorator]
@task_manager.lock_task(LOCK_NAME)  # type: ignore[untyped-decorator]
def periodic_sync_blackhole() -> None:
    """Define a huey periodic task."""
    with ctx:
        blackhole_service(False, True)

task_periodic_sync_seedbox

Define a huey tasks for seedbox synchronization.

Functions:

periodic_sync_seedbox –

Define a huey periodic task.

periodic_sync_seedbox

periodic_sync_seedbox() -> None

Define a huey periodic task.

Source code in seedboxsync/core/taskmanager/task/task_periodic_sync_seedbox.py

@task_manager.periodic_task(crontab(minute=minute), priority=PRIORITY)  # type: ignore[untyped-decorator]
@task_manager.lock_task(LOCK_NAME)  # type: ignore[untyped-decorator]
def periodic_sync_seedbox() -> None:
    """Define a huey periodic task."""
    with ctx:
        seedbox_service(False, True, False)

task_sync_blackhole

Define a huey tasks for blackhole synchronization.

Functions:

sync_blackhole –

Define a huey task.

sync_blackhole

sync_blackhole() -> None

Define a huey task.

Source code in seedboxsync/core/taskmanager/task/task_sync_blackhole.py

@task_manager.task(priority=PRIORITY)  # type: ignore[untyped-decorator]
@task_manager.lock_task(LOCK_NAME)  # type: ignore[untyped-decorator]
def sync_blackhole() -> None:
    """Define a huey task."""
    with ctx:
        blackhole_service(False, True)

task_sync_seedbox

Define a huey tasks for seedbox synchronization.

Functions:

sync_seedbox –

Define a huey task.

sync_seedbox

sync_seedbox() -> None

Define a huey task.

Source code in seedboxsync/core/taskmanager/task/task_sync_seedbox.py

@task_manager.task(priority=PRIORITY)  # type: ignore[untyped-decorator]
@task_manager.lock_task(LOCK_NAME)  # type: ignore[untyped-decorator]
def sync_seedbox() -> None:
    """Define a huey task."""
    with ctx:
        seedbox_service(False, True, False)

track_taskstatus

Track task status by decorator.

Functions:

heartbeat –

Task manager heartbeat.
heartbeat_shutdown –

Task manager heartbeat.
heartbeat_startup –

Task manager heartbeat.
track_taskstatus –

Track the execution status of a task in the database.

heartbeat

heartbeat() -> None

Task manager heartbeat.

Update the TaskStatus with key "heartbeat".

Source code in seedboxsync/core/taskmanager/track_taskstatus.py

def heartbeat() -> None:
    """
    Task manager heartbeat.

    Update the TaskStatus with key "heartbeat".
    """
    finished = datetime.now()
    TaskStatus.insert(
        key="heartbeat",
        finished=finished,
    ).on_conflict(
        conflict_target=[TaskStatus.key],
        update={
            TaskStatus.finished: finished,
        },
    ).execute()

heartbeat_shutdown

heartbeat_shutdown() -> None

Task manager heartbeat.

Update the TaskStatus with key "heartbeat".

Source code in seedboxsync/core/taskmanager/track_taskstatus.py

def heartbeat_shutdown() -> None:
    """
    Task manager heartbeat.

    Update the TaskStatus with key "heartbeat".
    """
    finished = datetime.now()
    TaskStatus.insert(
        key="heartbeat",
        running=False,
        finished=finished,
    ).on_conflict(
        conflict_target=[TaskStatus.key],
        update={
            TaskStatus.running: False,
            TaskStatus.finished: finished,
        },
    ).execute()

heartbeat_startup

heartbeat_startup() -> None

Task manager heartbeat.

Update the TaskStatus with key "heartbeat".

Source code in seedboxsync/core/taskmanager/track_taskstatus.py

def heartbeat_startup() -> None:
    """
    Task manager heartbeat.

    Update the TaskStatus with key "heartbeat".
    """
    started = datetime.now()
    TaskStatus.insert(
        key="heartbeat",
        running=True,
        started=started,
        finished=started,
    ).on_conflict(
        conflict_target=[TaskStatus.key],
        update={
            TaskStatus.running: True,
            TaskStatus.started: started,
            TaskStatus.finished: started,
        },
    ).execute()

track_taskstatus

track_taskstatus(key: str) -> Callable[[Callable[P, R]], Callable[P, R]]

Track the execution status of a task in the database.

Parameters:

key
(str) –

Unique identifier used to store the task status.

Returns:

Callable[[Callable[P, R]], Callable[P, R]] –

A decorator that updates the task status before and after execution.

Source code in seedboxsync/core/taskmanager/track_taskstatus.py

def track_taskstatus(key: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Track the execution status of a task in the database.

    Args:
        key: Unique identifier used to store the task status.

    Returns:
        A decorator that updates the task status before and after execution.
    """

    def decorator(function: Callable[P, R]) -> Callable[P, R]:
        @wraps(function)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            started = datetime.now()

            TaskStatus.insert(
                key=key,
                running=True,
                started=started,
                finished=None,
            ).on_conflict(
                conflict_target=[TaskStatus.key],
                update={
                    TaskStatus.running: True,
                    TaskStatus.started: started,
                    TaskStatus.finished: None,
                },
            ).execute()
            # Call heartbeat()
            heartbeat()

            try:
                return function(*args, **kwargs)
            finally:
                TaskStatus.update(
                    running=False,
                    finished=datetime.now(),
                ).where(
                    TaskStatus.key == key,
                ).execute()
                # Call heartbeat()
                heartbeat()

        return wrapper

    return decorator

utils

Task queue initialization and automatic task module registration.

Functions:

load_task_modules –

Import all task modules from this package.

load_task_modules

load_task_modules() -> list[ModuleType]

Import all task modules from this package.

Modules are loaded when their name starts with task_. Importing them registers their decorated Huey tasks in the task queue registry.

Returns:

list[ModuleType] –

The list of imported task modules.

Source code in seedboxsync/core/taskmanager/utils.py

def load_task_modules() -> list[ModuleType]:
    """
    Import all task modules from this package.

    Modules are loaded when their name starts with ``task_``. Importing them
    registers their decorated Huey tasks in the task queue registry.

    Returns:
        The list of imported task modules.
    """
    imported_modules: list[ModuleType] = []

    for module_info in pkgutil.iter_modules(task_package.__path__):
        if not module_info.name.startswith("task_"):
            continue

        module_name = f"{task_package.__name__}.{module_info.name}"
        current_app.logger.debug("Register task on %s", module_name)

        module = importlib.import_module(module_name)
        imported_modules.append(module)

    return imported_modules

utils

A collection of utility functions for SeedboxSync.

Functions:

byte_to_gi –

Convert in human readable units.
ensure_dir_exists –

Ensure the directory path exists, and if not create it.
get_database_path_from_paths –

Find and return an existing writable database file path.
get_mime_type_from_file –

Detect the MIME type and extension of a file.
get_torrent_infos –

Extracts information from a torrent file.
get_web_healthcheck_url –

Return the URL used to check the local Flask application.
is_running_in_docker –

Return whether the current process appears to run inside Docker.

byte_to_gi

byte_to_gi(bytes_value: float, suffix: str = 'B') -> str

Convert in human readable units.

Parameters:

bytes_value
(integer) –

Value not human readable.
suffix
(str, default: 'B' ) –

Suffix for value given to (default: B).

Returns:

str ( str ) –

human readable value in Gi.

Source code in seedboxsync/core/utils.py

def byte_to_gi(bytes_value: float, suffix: str = "B") -> str:
    """
    Convert in human readable units.

    Args:
        bytes_value (integer): Value not human readable.
        suffix (str): Suffix for value given to (default: B).

    Returns:
        str: human readable value in Gi.
    """
    gib = bytes_value / (1024**3)
    return f"{gib:.1f}Gi{suffix}"

ensure_dir_exists

ensure_dir_exists(path: str | PathLike[str]) -> None

Ensure the directory path exists, and if not create it.

Parameters:

path
(str) –

The filesystem path of a directory.

Raises:

AssertionError –

If the directory path exists, but is not a directory.

Source code in seedboxsync/core/utils.py

def ensure_dir_exists(path: str | PathLike[str]) -> None:
    """
    Ensure the directory ``path`` exists, and if not create it.

    Args:
        path (str): The filesystem path of a directory.

    Raises:
        AssertionError: If the directory ``path`` exists, but is not a directory.

    """
    path = Path(path).expanduser().resolve()

    if path.exists() and not path.is_dir():
        raise AssertionError(f"Path `{path}` exists but is not a directory!")
    if not path.exists():
        path.mkdir()

get_database_path_from_paths

get_database_path_from_paths() -> Path

Find and return an existing writable database file path.

Iterates through a predefined list of standard locations to locate a valid database file. Returns the first path that exists, is a regular file, and has write permissions. If no match is found, falls back to the default user configuration path.

Returns:

Path ( Path ) –

The resolved writable database path if found; otherwise, the default path (~/.config/seedboxsync/seedboxsync.db).

Source code in seedboxsync/core/utils.py

def get_database_path_from_paths() -> Path:
    """
    Find and return an existing writable database file path.

    Iterates through a predefined list of standard locations to locate a valid
    database file. Returns the first path that exists, is a regular file, and
    has write permissions. If no match is found, falls back to the default
    user configuration path.

    Returns:
        Path: The resolved writable database path if found; otherwise, the default
            path (~/.config/seedboxsync/seedboxsync.db).
    """
    db_paths = [
        Path("~/.config/seedboxsync/seedboxsync.db").expanduser().resolve(),
        Path("~/.seedboxsync.db").expanduser().resolve(),
        Path("~/.seedboxsync/config/seedboxsync.db").expanduser().resolve(),
        Path("/etc/seedboxsync/seedboxsync.db"),
    ]
    for path in db_paths:
        if path.exists() and path.is_file() and os.access(path, os.W_OK):
            return path
    return db_paths[0]

get_mime_type_from_file

get_mime_type_from_file(filename: str) -> tuple[str, str, str]

Detect the MIME type and extension of a file.

Parameters:

filename
(str) –

Name to the local file to analyze

Returns: tuple[str, str, str]: (mime_type, mime_extension, confidence)

Source code in seedboxsync/core/utils.py

def get_mime_type_from_file(filename: str) -> tuple[str, str, str]:
    """
    Detect the MIME type and extension of a file.

    Args:
        filename: Name to the local file to analyze
    Returns:
        tuple[str, str, str]: (mime_type, mime_extension, confidence)
    """
    # Initialize local file path
    local_filepath = Path(current_app.seedboxsync_config.get("local_download_path", "")).expanduser().resolve() / filename  # type: ignore[attr-defined]

    # Use first magic_file
    try:
        current_app.logger.debug(f"Attempting MIME detection via puremagic header analysis for: {local_filepath}")
        results = puremagic.magic_file(local_filepath)
        if results:
            match = results[0]

            mime_extension = match.extension.lstrip(".")
            mime_type = match.mime_type
            mime_confidence = f"puremagic (confidence: {match.confidence})"
            current_app.logger.debug(f"Successfully detected MIME via puremagic: type={mime_type}, ext={mime_extension}, confidence={mime_confidence}")

            return mime_type, mime_extension, mime_confidence

    except (FileNotFoundError, puremagic.PureError):
        pass

    # Fallback with mimetypes
    current_app.logger.debug(f"Attempting MIME detection fallback via mimetypes for: {local_filepath}")
    mime_type, _ = mimetypes.guess_type(local_filepath)

    if mime_type:
        extension = mimetypes.guess_extension(mime_type) or ""
        mime_extension = extension.lstrip(".")
        mime_confidence = "mimetypes (path_fallback)"

        current_app.logger.debug(f"MIME detection completed using fallback: type={mime_type}, ext={mime_extension}, confidence={mime_confidence}")

        return mime_type, mime_extension, "mimetypes (path_fallback)"

    # Last resort when nothing can be detected.
    parts = str(local_filepath).rsplit(".", 1) if local_filepath else []
    mime_extension = parts[1].lower() if len(parts) > 1 else "unknown"

    return "application/octet-stream", mime_extension, "unknown"

get_torrent_infos

get_torrent_infos(torrent_path: str | PathLike[str]) -> Any | None

Extracts information from a torrent file.

Parameters:

torrent_path
(str | PathLike[str]) –

Path to the torrent file.

Returns:

str ( Any | None ) –

Decoded torrent information.

Raises:

Exception –

If the file is not a valid torrent.

Source code in seedboxsync/core/utils.py

def get_torrent_infos(torrent_path: str | PathLike[str]) -> Any | None:
    """
    Extracts information from a torrent file.

    Args:
        torrent_path (str | PathLike[str]): Path to the torrent file.

    Returns:
        str: Decoded torrent information.

    Raises:
        Exception: If the file is not a valid torrent.
    """
    with Path(torrent_path).open("rb") as torrent:
        torrent_info = None

        try:
            torrent_info = bdecode(torrent.read())
        except Exception:
            current_app.logger.exception("Not valid torrent")
        finally:
            torrent.close()

        return torrent_info

get_web_healthcheck_url

get_web_healthcheck_url() -> str

Return the URL used to check the local Flask application.

Returns:

str ( str ) –

The healthcheck URL.

Source code in seedboxsync/core/utils.py

def get_web_healthcheck_url() -> str:
    """
    Return the URL used to check the local Flask application.

    Returns:
        str: The healthcheck URL.
    """
    explicit_url = os.getenv("HEALTHCHECK_URL")
    if explicit_url:
        return explicit_url.rstrip("/") + "/healthcheck"

    bind = os.getenv("BIND")
    if bind:
        return _healthcheck_url_from_bind(bind)

    port = 8000 if is_running_in_docker() else 5000
    return f"http://127.0.0.1:{port}/healthcheck"

is_running_in_docker

is_running_in_docker() -> bool

Return whether the current process appears to run inside Docker.

Returns:

bool ( bool ) –

True if in docker envoronment.

Source code in seedboxsync/core/utils.py

def is_running_in_docker() -> bool:
    """
    Return whether the current process appears to run inside Docker.

    Returns:
        bool: True if in docker envoronment.

    """
    # Test mountinfo
    if Path("/proc/self/mountinfo").exists():
        with Path("/proc/self/mountinfo").open() as f:
            if "docker" in f.read() or "overlay" in f.read():
                return True

    # Test du cgroup
    if Path("/proc/1/cgroup").exists():
        with Path("/proc/1/cgroup").open() as f:
            lines = f.read()
            if "docker" in lines or "kubepods" in lines:
                return True

    # Test /.dockerenv but not on podman
    return Path("/.dockerenv").exists()

taskmanager

Starter for the SeedboxSync taskmanager.

Functions:

on_shutdown –

Call function on huey shutdown.
on_startup –

Call function on huey startup.
reload_config –

Reload Flask DB config.
setup_worker_logging –

Configure the Huey logger from Flask.

on_shutdown

on_shutdown() -> None

Call function on huey shutdown.

Source code in seedboxsync/taskmanager.py

@huey.on_shutdown()  # type: ignore[untyped-decorator]
def on_shutdown() -> None:
    """Call function on huey shutdown."""
    heartbeat_shutdown()

on_startup

on_startup() -> None

Call function on huey startup.

Source code in seedboxsync/taskmanager.py

@huey.on_startup()  # type: ignore[untyped-decorator]
def on_startup() -> None:
    """Call function on huey startup."""
    heartbeat_startup()
    app.logger.debug("Flushing old tasks from queue...")
    huey.flush()

reload_config

reload_config(task: Any) -> None

Reload Flask DB config.

Source code in seedboxsync/taskmanager.py

@huey.pre_execute()  # type: ignore[untyped-decorator]
def reload_config(task: Any) -> None:
    """Reload Flask DB config."""
    config = Config.reload_config(app)
    app.config.from_mapping(config)

setup_worker_logging

setup_worker_logging(task: Any) -> None

Configure the Huey logger from Flask.

Source code in seedboxsync/taskmanager.py

@huey.pre_execute()  # type: ignore[untyped-decorator]
def setup_worker_logging(task: Any) -> None:
    """Configure the Huey logger from Flask."""
    huey_logger = logging.getLogger("huey")
    huey_logger.handlers = []
    for handler in app.logger.handlers:
        huey_logger.addHandler(handler)

seedboxsync

create_app

injected_config

__version__

app

cli

Cli

*args

**kwargs

command

*args

**kwargs

get_command

ctx

name

group

*args

**kwargs

invoke

ctx

list_commands

ctx

parse_args

ctx

args

Context

app cached property

render

data

headers

title

command

*args

**kwargs

group

*args

**kwargs

pass_context

func

cli

Cli

*args

**kwargs

command

*args

**kwargs

get_command

ctx

name

group

*args

**kwargs

invoke

ctx

list_commands

ctx

parse_args

ctx

args

Command

Group

command

*args

**kwargs

group

*args

**kwargs

check_root_warning

command

*args

**kwargs

group

*args

**kwargs

pass_context

func

commands

cmd_clean

cli

downloaded

`injected_config`

version

**`*args`**

`kwargs`**

**`*args`**

`kwargs`**

`ctx`

`name`

**`*args`**

`kwargs`**

`ctx`

`ctx`

`ctx`

`args`

app `cached` `property`

`data`

`headers`

`title`

**`*args`**

`kwargs`**

**`*args`**

`kwargs`**

`func`

**`*args`**

`kwargs`**

**`*args`**

`kwargs`**

`ctx`

`name`

**`*args`**

`kwargs`**

`ctx`

`ctx`

`ctx`

`args`

**`*args`**

`kwargs`**

**`*args`**

`kwargs`**

**`*args`**

`kwargs`**

**`*args`**

`kwargs`**

`func`

`id`

`ctx`

`ctx`

`number`

`search`

`ctx`

`number`

`search`

`ctx`

`number`

`search`

`ctx`

`ctx`

`ctx`

`ctx`

`ctx`

`dry_run`

`ping`

`ctx`

`dry_run`

`ping`

`only_store`

`ctx`

`username`

`email`

`password`

`ctx`

`id`

`yes`

`ctx`

`id`

`username`

`email`

`password`

`ctx`

`number`