aboutsummaryrefslogtreecommitdiff
path: root/webmentions_ssg/__init__.py
blob: b7996781678fedda4c8ca12ac202a5338d7c592e (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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
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
# SPDX-FileCopyrightText: 2026 Dennis Fink
#
# SPDX-License-Identifier: BSD-3-Clause

import logging
import logging.handlers
import os
from pathlib import Path

import click
import coloredlogs
import sqlalchemy as sa
from flask import Flask
from flask_bootstrap import Bootstrap5
from flask_migrate import Migrate, upgrade
from flask_sqlalchemy import SQLAlchemy
from flask_wtf.csrf import CSRFProtect
from sqlalchemy.orm import DeclarativeBase

from webmentions_ssg.config import DefaultConfig, DevelopmentConfig, ProductionConfig

from .tasks import Huey

APP_NAME = "webmentions-ssg"
VERSION = "0.0.1"


class Base(DeclarativeBase):
    pass


CSRF = CSRFProtect()
BOOTSTRAP = Bootstrap5()
DATABASE = SQLAlchemy(model_class=Base)
MIGRATE = Migrate()
HUEY = Huey()


def create_app(config_class: type[DefaultConfig] | None = None) -> Flask:
    """Creates and configures the Flask application."""

    development = os.environ.get(
        "WEBMENTIONS_SSG_DEV",
        "",
    ).lower() in {
        "1",
        "true",
        "yes",
        "on",
    }

    if config_class is None:
        config_class = DevelopmentConfig if development else ProductionConfig

    app = Flask(
        __name__,
        instance_path=str(Path("testing").absolute()) if development else None,
    )
    app.config.from_object(config_class)

    if not app.testing:
        app.config.from_pyfile(app.config["CONFIG_FILE"], silent=True)
        app.config.from_prefixed_env(prefix="FLASK")
        app.config.from_prefixed_env(prefix="WEBMENTIONS_SSG")

    default_logging_formatter = logging.Formatter(app.config["LOG_FORMAT"])

    log_file_handler = logging.handlers.RotatingFileHandler(
        app.config["LOG_FILE"],
        maxBytes=app.config["LOGFILE_MAX_BYTES"],
        backupCount=app.config["LOGFILE_BACKUP_COUNT"],
    )
    log_file_handler.setLevel(app.config["LOG_LEVEL"])
    log_file_handler.setFormatter(default_logging_formatter)

    logging.basicConfig(level=logging.DEBUG, handlers=[logging.NullHandler()])

    app.logger.handlers.clear()
    app.logger.setLevel(app.config["LOG_LEVEL"])
    app.logger.addHandler(log_file_handler)

    werkzeug_logger = logging.getLogger("werkzeug")
    werkzeug_logger.setLevel(app.config["LOG_LEVEL"])
    werkzeug_logger.addHandler(log_file_handler)

    coloredlogs.install(
        level=app.config["LOG_LEVEL"],
        logger=app.logger,
        fmt=app.config["LOG_FORMAT"],
    )
    coloredlogs.install(
        level=app.config["LOG_LEVEL"],
        logger=werkzeug_logger,
        fmt=app.config["LOG_FORMAT"],
    )

    DATABASE.init_app(app)
    MIGRATE.init_app(app, db=DATABASE)
    CSRF.init_app(app)

    from .auth import AUTH

    AUTH.init_app(app)
    BOOTSTRAP.init_app(app)
    HUEY.init_app(app)

    @app.context_processor
    def inject_version():
        return dict(VERSION=VERSION)

    from .views import root_page

    app.register_blueprint(root_page)

    @app.cli.command("create-db")
    def create_db() -> None:
        """Create or update the database to the latest migration."""
        upgrade()

    @app.cli.command("create-user")
    @click.argument("username")
    @click.password_option(
        confirmation_prompt=True,
    )
    def create_user(
        username: str,
        password: str,
    ) -> None:
        """Create a user."""

        from .models import User

        if (
            DATABASE.session.scalar(sa.select(User).where(User.username == username))
            is not None
        ):
            raise click.ClickException(f"User {username!r} already exists.")

        user = User(
            username=username,
        )
        user.set_password(password)

        DATABASE.session.add(user)
        DATABASE.session.commit()

        click.echo(f"Created user {username!r}.")

    return app