Skip to content

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 –

    Seedbox Core package.

  • taskmanager –

    Starter for the SeedboxSync taskmanager.

Functions:

  • create_app –

    Create and configure the SeedboxSync Flask application.

create_app 

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

Create and configure the SeedboxSync Flask application.

Parameters:

  • test_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 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def create_app(test_config: dict[str, str] | None = None) -> Flask:
    """
    Create and configure the SeedboxSync Flask application.

    Args:
        test_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,
    )

    # Configure logger for Flask and Click
    logger.configure_logger(app.logger)

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

    # Initialize the database
    Database(app)

    # Load config
    Config(app)

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

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

    # Inject Jinja function / variables
    @app.context_processor
    def inject_formatters() -> dict[str, Callable[[datetime], str]]:  # pyright: ignore [reportUnusedFunction]
        return {"format_datetime": format_datetime}

    @app.context_processor
    def inject_globals() -> dict[str, str]:  # pyright: ignore [reportUnusedFunction]
        locale = get_babel_locale() or app.config.get("BABEL_DEFAULT_LOCALE", "en")
        theme = app.config.get(Config.CONFIG_NAMESPACE + "WEBUI_THEME", "auto")
        return {
            "version": version,
            "theme": theme,
            "api_version": api_version,
            "locale": str(locale).replace("_", "-"),
        }

    # Initialize the cache
    cache.init_app(app)

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

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

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

    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 
53
54
55
56
57
58
59
60
61
62
63
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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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 
109
110
111
112
113
114
115
116
117
118
119
120
121
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 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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 
82
83
84
85
86
87
88
89
90
91
92
93
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:

Methods:

  • render –

    Render tabular data using the specified output format.

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: str | dict[str, str] | Sequence[str],
    tablefmt: str = "github",
) -> Any

Render tabular data using the specified output format.

Parameters:

  • data
    (Iterable[Any]) –

    Tabular data to render.

  • headers
    (str | dict[str, str] | Sequence[str]) –

    Column headers passed to tabulate.

  • tablefmt
    (str, default: 'github' ) –

    Output table format.

Returns:

  • str ( Any ) –

    The formatted table.

Source code in seedboxsync/cli/context.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
def render(self, data: Iterable[Any], headers: str | dict[str, str] | Sequence[str], tablefmt: str = "github") -> Any:
    """
    Render tabular data using the specified output format.

    Args:
        data: Tabular data to render.
        headers: Column headers passed to ``tabulate``.
        tablefmt: Output table format.

    Returns:
        str: The formatted table.
    """
    return tabulate(data, headers=headers, tablefmt=tablefmt)

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 
218
219
220
221
222
223
224
225
226
227
228
229
230
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 
233
234
235
236
237
238
239
240
241
242
243
244
245
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 
18
19
20
21
22
23
24
25
26
27
28
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 
53
54
55
56
57
58
59
60
61
62
63
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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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 
109
110
111
112
113
114
115
116
117
118
119
120
121
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 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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 
82
83
84
85
86
87
88
89
90
91
92
93
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 
189
190
191
192
193
194
195
196
197
198
199
200
201
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 
203
204
205
206
207
208
209
210
211
212
213
214
215
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 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
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 
218
219
220
221
222
223
224
225
226
227
228
229
230
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 
233
234
235
236
237
238
239
240
241
242
243
244
245
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 
18
19
20
21
22
23
24
25
26
27
28
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_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 
13
14
15
@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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@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 
18
19
20
21
22
23
24
25
26
27
28
29
@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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@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
    db_version = SeedboxSync.get_db_version()
    click.echo(f"Version: {version}")
    click.echo(f"Database version: {db_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)

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 
15
16
17
18
@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 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@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"),
            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",
                "finished": "Finished",
                "path": "Path",
                "size": "Size",
            },
            tablefmt="github",
        )
    )
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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@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 = typed_peewee_dicts(
        Download.select(
            Download.id,
            fn.SUBSTR(Download.path, -100).alias("path"),
            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()
    )

    # Define the output column order explicitly
    rows = [
        (
            row["id"],
            row["path"],
            row["started"],
            row["progress"],
            row["eta"],
            row["size"],
        )
        for row in reversed(list(data))
    ]

    click.echo(
        ctx.render(
            rows,
            headers=["Id", "Path", "Started", "Progress", "ETA", "Size"],
            tablefmt="github",
        )
    )
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 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@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
    where = Torrent.name.contains(search) if search else ~Torrent.id.contains("not_a_int")

    # DB query
    data = Torrent.select(Torrent.id, Torrent.name, Torrent.sent).where(where).limit(number).order_by(Torrent.sent.desc()).dicts()

    click.echo(
        ctx.render(
            reversed(data),
            headers={"id": "Id", "name": "Name", "sent": "Sent datetime"},
            tablefmt="github",
        )
    )

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 
36
37
38
39
40
41
42
43
44
45
@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 
48
49
50
51
52
53
54
55
56
57
@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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@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 
20
21
22
@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 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@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 
cli() -> None

Empty function for Click sub commands.

Source code in seedboxsync/cli/commands/cmd_task.py 
14
15
16
@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 
55
56
57
58
59
60
@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 
63
64
65
66
67
68
@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 
43
44
45
46
47
48
49
50
51
52
@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 = []
    for task in ctx.app.task_manager._registry._registry:
        data.append([task])

    click.echo(ctx.render(data, headers=["Class"]))
tasks_pending 
tasks_pending(ctx: Context) -> None

List pending tasks.

Source code in seedboxsync/cli/commands/cmd_task.py 
30
31
32
33
34
35
36
37
38
39
40
@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():
        name = getattr(task, "name", str(task).split(": ")[0])
        task_id = getattr(task, "id", str(task).split(": ")[-1] if ": " in str(task) else str(task))
        data.append([name, task_id])

    click.echo(ctx.render(data, headers=["Task Name", "Task ID / UUID"]))
tasks_result 
tasks_result(ctx: Context) -> None

List all_results() tasks.

Source code in seedboxsync/cli/commands/cmd_task.py 
19
20
21
22
23
24
25
26
27
@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 = []
    for task in ctx.app.task_manager.all_results():
        data.append([task])

    click.echo(ctx.render(data, headers=["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 
71
72
73
74
75
76
77
78
79
@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 
82
83
84
85
86
87
88
89
90
@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")

context

Build a context used by Click.

Classes:

  • Context –

    SeedboxSync Click context.

Context 


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

              

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

SeedboxSync Click context.

Used by:

Methods:

  • render –

    Render tabular data using the specified output format.

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: str | dict[str, str] | Sequence[str],
    tablefmt: str = "github",
) -> Any

Render tabular data using the specified output format.

Parameters:

  • data
    (Iterable[Any]) –

    Tabular data to render.

  • headers
    (str | dict[str, str] | Sequence[str]) –

    Column headers passed to tabulate.

  • tablefmt
    (str, default: 'github' ) –

    Output table format.

Returns:

  • str ( Any ) –

    The formatted table.

Source code in seedboxsync/cli/context.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
def render(self, data: Iterable[Any], headers: str | dict[str, str] | Sequence[str], tablefmt: str = "github") -> Any:
    """
    Render tabular data using the specified output format.

    Args:
        data: Tabular data to render.
        headers: Column headers passed to ``tabulate``.
        tablefmt: Output table format.

    Returns:
        str: The formatted table.
    """
    return tabulate(data, headers=headers, tablefmt=tablefmt)

core

Seedbox Core package.

Modules:

  • config –

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

  • dao –

    DAO package with all Peewee models.

  • db –

    Database module.

  • 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 connector using peewee.

  • 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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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)

reload_config staticmethod 

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

Reload application configuration from the database.

Source code in seedboxsync/core/config.py 
141
142
143
144
@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 connector using peewee.

Attributes:

  • app (Flask) –

    The Flask application that owns the database connection.

Parameters:

  • app

    (Flask) –

    The Flask application to bind to the database.

Methods:

  • migrate_to_2 –

    Migration: rebuild SeedboxSync table and add Lock table.

  • migrate_to_3 –

    Migration: allow null values for the 'announce' field in the torrent table.

  • migrate_to_4 –

    Replace 'Lock' table by 'TaskStatus'.

Source code in seedboxsync/core/db.py 
39
40
41
42
43
44
45
46
47
48
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()

migrate_to_2 

migrate_to_2() -> None

Migration: rebuild SeedboxSync table and add Lock table.

Fixes compatibility issues between tables created with Peewee v2 and v3.

Source code in seedboxsync/core/db.py 
140
141
142
143
144
145
146
147
148
def migrate_to_2(self) -> None:
    """
    Migration: rebuild SeedboxSync table and add Lock table.

    Fixes compatibility issues between tables created with Peewee v2 and v3.
    """
    self.db.drop_tables([SeedboxSync])
    self.db.create_tables([SeedboxSync])
    SeedboxSync.set_db_version("2")

migrate_to_3 

migrate_to_3() -> None

Migration: allow null values for the 'announce' field in the torrent table.

Source code in seedboxsync/core/db.py 
150
151
152
153
154
155
156
def migrate_to_3(self) -> None:
    """Migration: allow null values for the 'announce' field in the torrent table."""
    migrator = SchemaMigrator.from_database(self.db)
    migrate(
        migrator.drop_not_null("torrent", "announce"),
    )
    SeedboxSync.set_db_version("3")

migrate_to_4 

migrate_to_4() -> None

Replace 'Lock' table by 'TaskStatus'.

Source code in seedboxsync/core/db.py 
158
159
160
161
162
163
164
def migrate_to_4(self) -> None:
    """Replace 'Lock' table by 'TaskStatus'."""
    self.db.execute_sql("DROP TABLE IF EXISTS lock;")  # type: ignore[no-untyped-call]
    self.db.execute_sql("DELETE FROM seedboxsync WHERE key = 'version';")  # type: ignore[no-untyped-call]
    self.db.create_tables([TaskStatus])

    SeedboxSync.set_db_version("4")

Flask 


              flowchart TD
              seedboxsync.core.Flask[Flask]

              

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

Flask application with SeedboxSync-specific configuration helpers.

Returned by:

Attributes:

ping cached property 

ping: AbstractPingClient

Return the configured ping client instance.

Returns:

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:

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(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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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)
reload_config staticmethod 
reload_config(app: Flask) -> dict[str, str]

Reload application configuration from the database.

Source code in seedboxsync/core/config.py 
141
142
143
144
@staticmethod
def reload_config(app: Flask) -> dict[str, str]:
    """Reload application configuration from the database."""
    return Config._load_config_from_database(app)

dao

DAO package with all Peewee models.

Modules:

  • 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.

Classes:

  • 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.

Download 


              flowchart TD
              seedboxsync.core.dao.Download[Download]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.Download
                


              click seedboxsync.core.dao.Download href "" "seedboxsync.core.dao.Download"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.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:

Methods:

is_already_download staticmethod 
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/dao/download.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@staticmethod
def is_already_download(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 = Download.select().where(Download.path == filepath, Download.finished > 0).count()
    return count != 0

SeedboxSync 


              flowchart TD
              seedboxsync.core.dao.SeedboxSync[SeedboxSync]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.SeedboxSync
                


              click seedboxsync.core.dao.SeedboxSync href "" "seedboxsync.core.dao.SeedboxSync"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.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.

Methods:

get_db_version staticmethod 
get_db_version() -> str

Return database model version.

Returns:

  • str ( str ) –

    The database model version.

Source code in seedboxsync/core/dao/seedboxsync.py 
29
30
31
32
33
34
35
36
37
@staticmethod
def get_db_version() -> str:
    """
    Return database model version.

    Returns:
        str: The database model version.
    """
    return str(SeedboxSync.select(SeedboxSync.value).where(SeedboxSync.key == "db_version").first().value)  # Use old style to prevent Peewee 2 databases.
set_db_version staticmethod 
set_db_version(db_version: str) -> None

Upsert database model version.

Parameters:

  • db_version
    (str) –

    The database model version.

Source code in seedboxsync/core/dao/seedboxsync.py 
39
40
41
42
43
44
45
46
47
@staticmethod
def set_db_version(db_version: str) -> None:
    """
    Upsert database model version.

    Args:
        db_version (str): The database model version.
    """
    SeedboxSync.replace(key="db_version", value=db_version).execute()  # type: ignore[no-untyped-call]

SeedboxSyncModel 


              flowchart TD
              seedboxsync.core.dao.SeedboxSyncModel[SeedboxSyncModel]

              

              click seedboxsync.core.dao.SeedboxSyncModel href "" "seedboxsync.core.dao.SeedboxSyncModel"

Basemodel from which all other peewee models are derived.

TaskStatus 


              flowchart TD
              seedboxsync.core.dao.TaskStatus[TaskStatus]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.TaskStatus
                


              click seedboxsync.core.dao.TaskStatus href "" "seedboxsync.core.dao.TaskStatus"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.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.dao.Torrent[Torrent]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.Torrent
                


              click seedboxsync.core.dao.Torrent href "" "seedboxsync.core.dao.Torrent"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.model.SeedboxSyncModel"

Data Access Object (DAO) representing a torrent.

This model stores basic metadata for a torrent, including its name, announce URL, and the timestamp when it was sent or added to the system.

Attributes:

  • id (int) –

    Auto-incremented primary key.

  • name (str) –

    Name of the torrent.

  • announce (str) –

    Tracker announce URL.

  • sent (datetime) –

    Timestamp indicating when the torrent was sent or created.

download

Peewee DAO model for Download.

Classes:

  • Download –

    Data Access Object (DAO) representing a file download.

Download 

              flowchart TD
              seedboxsync.core.dao.download.Download[Download]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.download.Download
                


              click seedboxsync.core.dao.download.Download href "" "seedboxsync.core.dao.download.Download"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.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:

Methods:

is_already_download staticmethod 
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/dao/download.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@staticmethod
def is_already_download(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 = Download.select().where(Download.path == filepath, Download.finished > 0).count()
    return count != 0

model

Peewee model.

Classes:

  • SeedboxSyncModel –

    Basemodel from which all other peewee models are derived.

SeedboxSyncModel 

              flowchart TD
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

              

              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.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.dao.seedboxsync.SeedboxSync[SeedboxSync]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.seedboxsync.SeedboxSync
                


              click seedboxsync.core.dao.seedboxsync.SeedboxSync href "" "seedboxsync.core.dao.seedboxsync.SeedboxSync"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.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.

Methods:

get_db_version staticmethod 
get_db_version() -> str

Return database model version.

Returns:

  • str ( str ) –

    The database model version.

Source code in seedboxsync/core/dao/seedboxsync.py 
29
30
31
32
33
34
35
36
37
@staticmethod
def get_db_version() -> str:
    """
    Return database model version.

    Returns:
        str: The database model version.
    """
    return str(SeedboxSync.select(SeedboxSync.value).where(SeedboxSync.key == "db_version").first().value)  # Use old style to prevent Peewee 2 databases.
set_db_version staticmethod 
set_db_version(db_version: str) -> None

Upsert database model version.

Parameters:

  • db_version (str) –

    The database model version.

Source code in seedboxsync/core/dao/seedboxsync.py 
39
40
41
42
43
44
45
46
47
@staticmethod
def set_db_version(db_version: str) -> None:
    """
    Upsert database model version.

    Args:
        db_version (str): The database model version.
    """
    SeedboxSync.replace(key="db_version", value=db_version).execute()  # type: ignore[no-untyped-call]

taskstatus

Peewee DAO model for TaskStatus.

Classes:

  • TaskStatus –

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

TaskStatus 

              flowchart TD
              seedboxsync.core.dao.taskstatus.TaskStatus[TaskStatus]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.taskstatus.TaskStatus
                


              click seedboxsync.core.dao.taskstatus.TaskStatus href "" "seedboxsync.core.dao.taskstatus.TaskStatus"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.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.dao.torrent.Torrent[Torrent]
              seedboxsync.core.dao.model.SeedboxSyncModel[SeedboxSyncModel]

                              seedboxsync.core.dao.model.SeedboxSyncModel --> seedboxsync.core.dao.torrent.Torrent
                


              click seedboxsync.core.dao.torrent.Torrent href "" "seedboxsync.core.dao.torrent.Torrent"
              click seedboxsync.core.dao.model.SeedboxSyncModel href "" "seedboxsync.core.dao.model.SeedboxSyncModel"

Data Access Object (DAO) representing a torrent.

This model stores basic metadata for a torrent, including its name, announce URL, and the timestamp when it was sent or added to the system.

Attributes:

  • id (int) –

    Auto-incremented primary key.

  • name (str) –

    Name of the torrent.

  • announce (str) –

    Tracker announce URL.

  • sent (datetime) –

    Timestamp indicating when the torrent was sent or created.

db

Database module.

Classes:

  • Database –

    Database connector using peewee.

Database 

Database(app: Flask)

Database connector using peewee.

Attributes:

  • app (Flask) –

    The Flask application that owns the database connection.

Parameters:

  • app
    (Flask) –

    The Flask application to bind to the database.

Methods:

  • migrate_to_2 –

    Migration: rebuild SeedboxSync table and add Lock table.

  • migrate_to_3 –

    Migration: allow null values for the 'announce' field in the torrent table.

  • migrate_to_4 –

    Replace 'Lock' table by 'TaskStatus'.

Source code in seedboxsync/core/db.py 
39
40
41
42
43
44
45
46
47
48
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()
migrate_to_2 
migrate_to_2() -> None

Migration: rebuild SeedboxSync table and add Lock table.

Fixes compatibility issues between tables created with Peewee v2 and v3.

Source code in seedboxsync/core/db.py 
140
141
142
143
144
145
146
147
148
def migrate_to_2(self) -> None:
    """
    Migration: rebuild SeedboxSync table and add Lock table.

    Fixes compatibility issues between tables created with Peewee v2 and v3.
    """
    self.db.drop_tables([SeedboxSync])
    self.db.create_tables([SeedboxSync])
    SeedboxSync.set_db_version("2")
migrate_to_3 
migrate_to_3() -> None

Migration: allow null values for the 'announce' field in the torrent table.

Source code in seedboxsync/core/db.py 
150
151
152
153
154
155
156
def migrate_to_3(self) -> None:
    """Migration: allow null values for the 'announce' field in the torrent table."""
    migrator = SchemaMigrator.from_database(self.db)
    migrate(
        migrator.drop_not_null("torrent", "announce"),
    )
    SeedboxSync.set_db_version("3")
migrate_to_4 
migrate_to_4() -> None

Replace 'Lock' table by 'TaskStatus'.

Source code in seedboxsync/core/db.py 
158
159
160
161
162
163
164
def migrate_to_4(self) -> None:
    """Replace 'Lock' table by 'TaskStatus'."""
    self.db.execute_sql("DROP TABLE IF EXISTS lock;")  # type: ignore[no-untyped-call]
    self.db.execute_sql("DELETE FROM seedboxsync WHERE key = 'version';")  # type: ignore[no-untyped-call]
    self.db.create_tables([TaskStatus])

    SeedboxSync.set_db_version("4")

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 

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 
33
34
35
36
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 
33
34
35
36
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 
33
34
35
36
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 
33
34
35
36
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 
33
34
35
36
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:

Attributes:

ping cached property 
ping: AbstractPingClient

Return the configured ping client instance.

Returns:

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:

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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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:

Classes:

AbstractPingClient 

AbstractPingClient()

Abstract base class for transport clients.

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

Returned by:

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 
27
28
29
@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 
31
32
33
34
35
36
37
38
@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 
40
41
42
43
44
45
46
47
@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 
AbstractPingClient()

Abstract base class for transport clients.

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

Returned by:

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 
27
28
29
@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 
31
32
33
34
35
36
37
38
@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 
40
41
42
43
44
45
46
47
@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.

Classes:

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 
16
17
18
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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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:

Classes:

AbstractSyncClient 

AbstractSyncClient()

Abstract base class for transport clients.

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

Returned by:

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 
34
35
36
@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 
77
78
79
80
81
82
83
84
@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 
86
87
88
89
90
91
92
93
94
95
96
@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 
108
109
110
@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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@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 
38
39
40
41
42
43
44
45
46
47
@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 
 98
 99
100
101
102
103
104
105
106
@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 
65
66
67
68
69
70
71
72
73
74
75
@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 
AbstractSyncClient()

Abstract base class for transport clients.

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

Returned by:

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 
34
35
36
@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 
77
78
79
80
81
82
83
84
@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 
86
87
88
89
90
91
92
93
94
95
96
@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 
108
109
110
@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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@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 
38
39
40
41
42
43
44
45
46
47
@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 
 98
 99
100
101
102
103
104
105
106
@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 
65
66
67
68
69
70
71
72
73
74
75
@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 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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 
176
177
178
179
180
181
182
183
184
185
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 
187
188
189
190
191
192
193
194
195
196
197
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 
251
252
253
254
255
256
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 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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 
109
110
111
112
113
114
115
116
117
118
119
120
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 
199
200
201
202
203
204
205
206
207
208
209
210
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 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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 
158
159
160
161
162
163
164
165
166
167
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 
169
170
171
172
173
174
175
176
177
178
179
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 
226
227
228
229
230
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 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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 
181
182
183
184
185
186
187
188
189
190
191
192
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 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
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 
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 
17
18
19
20
21
22
23
24
25
26
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 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@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 app.seedboxsync_config.get("sync_blackhole_enabled"):
        app.logger.info("Blackhole synchronization task is disabled")
        return

    app.logger.debug(f'sync blackhole dry-run: "{dry_run}"')
    app.logger.debug(f'sync blackhole ping: "{ping}"')

    # Call ping.start() if enabled
    if ping:
        app.ping.start("sync_blackhole")

    # Gather all torrent files
    local_watch_path = app.seedboxsync_config.get("local_watch_path", "")
    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:
        app.logger.info('No torrent files found in "{}"'.format(app.seedboxsync_config.get("local_watch_path")))
        # Call ping.success() if enabled
        if ping:
            app.ping.success("sync_blackhole")
        return

    for torrent_file in torrents:
        torrent_name = torrent_file.name

        # Dry-run mode
        if dry_run:
            app.logger.info(f'Dry-run: not uploading torrent "{torrent_name}"')
            continue

        tmp_path = app.seedboxsync_config.get("seedbox_tmp_path", "")
        watch_path = app.seedboxsync_config.get("seedbox_watch_path", "")

        app.logger.info(f'Upload torrent: "{torrent_name}"')
        app.logger.debug(f'Upload "{torrent_file}" to "{tmp_path}"')

        try:
            app.sync.chdir(None)  # type: ignore[arg-type]
            app.sync.put(torrent_file, Path(tmp_path) / torrent_name)

            # Apply chmod if configured
            chmod = app.seedboxsync_config.get("seedbox_chmod", False)
            if isinstance(chmod, str):
                app.logger.debug(f"Change permissions to {chmod}")
                app.sync.chmod(Path(tmp_path) / torrent_name, int(chmod, 8))

            # Move file from tmp to watch directory
            app.logger.debug(f'Move from "{tmp_path}" to "{watch_path}"')
            app.sync.rename(
                Path(tmp_path) / torrent_name,
                Path(watch_path) / torrent_name,
            )

            # Store torrent info in database
            torrent_info = utils.get_torrent_infos(torrent_file)
            torrent = Torrent.create(name=torrent_name)
            if torrent_info is not None and isinstance(torrent_info, dict):
                torrent.announce = torrent_info.get("announce")
                torrent.save()

                # Remove local torrent file
                app.logger.debug(f'Remove local torrent "{torrent_file}"')
                Path(torrent_file).unlink()
            else:
                app.logger.warning(f'Rename local "{torrent_file}" to .torrent.fail')
                Path(torrent_file).rename(fspath(torrent_file) + ".fail")
        except SSHException as exc:
            app.logger.warning(f"SSH client exception > {exc!s}")

    # Call ping.success() if enabled
    if ping:
        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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@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 app.seedboxsync_config.get("sync_seedbox_enabled"):
        app.logger.info("Seedbox synchronization task is disabled")
        return

    app.logger.debug(f'sync seedbox dry-run: "{dry_run}"')
    app.logger.debug(f'sync seeddbox only-store: "{only_store}"')
    app.logger.debug(f'sync seedbox ping: "{ping}"')

    # Call ping.start() if enabled
    if ping:
        app.ping.start("sync_seedbox")

    finished_path = app.seedboxsync_config.get("seedbox_finished_path", "")
    part_suffix = app.seedboxsync_config.get("seedbox_part_suffix")
    app.logger.debug(f'Scanning files in "{finished_path}"')

    # Walk through all files on the seedbox
    try:
        try:
            app.sync.chdir(None)  # type: ignore[arg-type]
            app.sync.chdir(finished_path)
        except FileNotFoundError as exc:
            app.logger.error(f"{exc!s}\nFailed to scan directory: {finished_path}")
            return

        for root, _, filenames in app.sync.walk(""):  # type: ignore[attr-defined]
            root = Path(root)

            for filename in filenames:
                filepath = root / filename

                if filepath.suffix == part_suffix:
                    app.logger.debug(f'Skipping part file "{filename}"')
                elif Download.is_already_download(filepath):
                    app.logger.debug(f'Skipping already downloaded file "{filename}"')
                elif __exclude_by_pattern(filepath):
                    app.logger.debug(f'Skipping excluded file "{filename}"')
                else:
                    if dry_run:
                        app.logger.info(f'Dry-run: not downloading "{filepath}"')
                        continue

                    __get_file(filepath, only_store)

    except (OSError, FileNotFoundError) as exc:
        app.logger.error(f'SeedboxSyncError > "{exc}"')

    # Call ping.success() if enabled
    if ping:
        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:

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:

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 
28
29
30
31
32
33
34
35
36
37
38
39
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 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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 
57
58
59
60
61
62
63
64
65
66
67
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 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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:

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 
28
29
30
31
32
33
34
35
36
37
38
39
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 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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 
57
58
59
60
61
62
63
64
65
66
67
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.

Functions:

periodic_sync_blackhole 
periodic_sync_blackhole() -> None

Define a huey periodic task.

Source code in seedboxsync/core/taskmanager/task/task_periodic_sync_blackhole.py 
23
24
25
26
27
28
@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 
periodic_sync_seedbox() -> None

Define a huey periodic task.

Source code in seedboxsync/core/taskmanager/task/task_periodic_sync_seedbox.py 
23
24
25
26
27
28
@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 
sync_blackhole() -> None

Define a huey task.

Source code in seedboxsync/core/taskmanager/task/task_sync_blackhole.py 
22
23
24
25
26
27
@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 
sync_seedbox() -> None

Define a huey task.

Source code in seedboxsync/core/taskmanager/task/task_sync_seedbox.py 
22
23
24
25
26
27
@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 
heartbeat() -> None

Task manager heartbeat.

Update the TaskStatus with key "heartbeat".

Source code in seedboxsync/core/taskmanager/track_taskstatus.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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 
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 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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}"
        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 

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 
18
19
20
21
22
23
24
25
26
27
28
29
30
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 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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_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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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:
            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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 

on_shutdown() -> None

Call function on huey shutdown.

Source code in seedboxsync/taskmanager.py 
32
33
34
35
@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 
25
26
27
28
29
30
@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 
45
46
47
48
49
@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 
37
38
39
40
41
42
43
@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)