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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
# 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.cli import AppGroup
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()
USERS_CLI = AppGroup("users", help="Handle users")
DB_CLI = AppGroup("db", help="Handle database")
def create_app(config_class: type[DefaultConfig] | None = None) -> Flask:
"""
Create and configure the Flask application.
The application configuration is selected based on the development
environment unless an explicit configuration class is provided. Logging,
extensions, authentication, Huey, blueprints, and CLI commands are then
initialized.
:param config_class: Configuration class to use, or ``None`` to select the
development or production configuration automatically.
:return: Configured 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():
"""
Add the application version to the template context.
:return: Template context containing the application version.
"""
return {"VERSION": VERSION}
from .views import root_page
app.register_blueprint(root_page)
app.cli.add_command(USERS_CLI)
return app
@USERS_CLI.command("create", help="Create or add users")
@click.option("--username", prompt=True)
@click.password_option(confirmation_prompt=True)
def create_user(username: str, password: str) -> None:
"""
Create a new user.
:param username: Username of the user to create.
:param password: Plain-text password to hash and store for the user.
:raises click.ClickException: If a user with the given username already
exists.
"""
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, password=password)
DATABASE.session.add(user)
DATABASE.session.commit()
click.echo(f"Created user {username!r}.")
@DB_CLI.command("create")
def create_db() -> None:
"""
Upgrade the database to the latest migration revision.
"""
upgrade()
|
