yannic/stacosys
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #18 from kianby/feature-refactor-startup

Browse source 
Feature refactor startup
This commit is contained in:
Yax 2024-04-12 21:38:33 +02:00 • committed by GitHub
commit ac5345deec
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 124 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
requirements-dev.lock
View file

@ -10,7 +10,7 @@
altgraph==0.17.4
astroid==3.1.0
background==0.2.1
black==24.2.0
black==24.3.0
blinker==1.7.0
certifi==2024.2.2
charset-normalizer==3.3.2
@ -20,34 +20,34 @@ coveralls==3.3.1
dill==0.3.8
docopt==0.6.2
exceptiongroup==1.2.0
flask==3.0.2
idna==3.6
flask==3.0.3
idna==3.7
iniconfig==2.0.0
isort==5.13.2
itsdangerous==2.1.2
jinja2==3.1.3
markdown==3.5.2
markdown==3.6
markupsafe==2.1.5
mccabe==0.7.0
mypy==1.8.0
mypy==1.9.0
mypy-extensions==1.0.0
packaging==23.2
packaging==24.0
pathspec==0.12.1
platformdirs==4.2.0
pluggy==1.4.0
pydal==20231114.3
pyinstaller==6.4.0
pyinstaller-hooks-contrib==2024.1
pyinstaller==6.5.0
pyinstaller-hooks-contrib==2024.3
pylint==3.1.0
pyrss2gen==1.1
pytest==8.0.2
pytest-cov==4.1.0
pytest==8.1.1
pytest-cov==5.0.0
requests==2.31.0
tomli==2.0.1
tomlkit==0.12.4
types-markdown==3.5.0.20240129
typing-extensions==4.10.0
types-markdown==3.6.0.20240316
typing-extensions==4.11.0
urllib3==2.2.1
werkzeug==3.0.1
werkzeug==3.0.2
# The following packages are considered to be unsafe in a requirements file:
setuptools==69.1.1
setuptools==69.3.0

10
requirements.lock
View file

@ -12,15 +12,15 @@ blinker==1.7.0
certifi==2024.2.2
charset-normalizer==3.3.2
click==8.1.7
flask==3.0.2
idna==3.6
flask==3.0.3
idna==3.7
itsdangerous==2.1.2
jinja2==3.1.3
markdown==3.5.2
markdown==3.6
markupsafe==2.1.5
pydal==20231114.3
pyrss2gen==1.1
requests==2.31.0
types-markdown==3.5.0.20240129
types-markdown==3.6.0.20240316
urllib3==2.2.1
werkzeug==3.0.1
werkzeug==3.0.2

7
src/stacosys/interface/__init__.py
View file

@ -7,7 +7,6 @@ import background
from flask import Flask
from stacosys.db import dao
from stacosys.service import config, mailer
from stacosys.service.configuration import ConfigParameter
app = Flask(__name__)
@ -20,7 +19,7 @@ logger = logging.getLogger(__name__)
@background.task
def submit_new_comment(comment):
site_url = config.get(ConfigParameter.SITE_URL)
site_url = app.config["CONFIG"].get(ConfigParameter.SITE_URL)
comment_list = (
f"Web admin interface: {site_url}/web/admin",
"",
@ -35,9 +34,9 @@ def submit_new_comment(comment):
email_body = "\n".join(comment_list)
# send email to notify admin
site_name = config.get(ConfigParameter.SITE_NAME)
site_name = app.config["CONFIG"].get(ConfigParameter.SITE_NAME)
subject = f"STACOSYS {site_name}"
if mailer.send(subject, email_body):
if app.config["MAILER"].send(subject, email_body):
logger.debug("new comment processed")
# save notification datetime
dao.notify_comment(comment)

3
src/stacosys/interface/form.py
View file

@ -6,7 +6,6 @@ from flask import abort, redirect, request
from stacosys.db import dao
from stacosys.interface import app, submit_new_comment
from stacosys.service import config
from stacosys.service.configuration import ConfigParameter
logger = logging.getLogger(__name__)
@ -47,7 +46,7 @@ def new_form_comment():
# send notification e-mail asynchronously
submit_new_comment(comment)
return redirect(config.get(ConfigParameter.SITE_REDIRECT), code=302)
return redirect(app.config["CONFIG"].get(ConfigParameter.SITE_REDIRECT), code=302)
def check_form_data(posted_comment):

17
src/stacosys/interface/web/admin.py
View file

@ -8,7 +8,6 @@ from flask import flash, redirect, render_template, request, session
from stacosys.db import dao
from stacosys.interface import app
from stacosys.service import config, rss
from stacosys.service.configuration import ConfigParameter
logger = logging.getLogger(__name__)
@ -25,8 +24,8 @@ def index():
def is_login_ok(username, password):
hashed = hashlib.sha256(password.encode()).hexdigest().upper()
return (
config.get(ConfigParameter.WEB_USERNAME) == username
and config.get(ConfigParameter.WEB_PASSWORD) == hashed
app.config["CONFIG"].get(ConfigParameter.WEB_USERNAME) == username
and app.config["CONFIG"].get(ConfigParameter.WEB_PASSWORD) == hashed
)
@ -42,7 +41,9 @@ def login():
flash("Identifiant ou mot de passe incorrect")
return redirect("/web/login")
# GET
return render_template("login_" + config.get(ConfigParameter.LANG) + ".html")
return render_template(
"login_" + app.config["CONFIG"].get(ConfigParameter.LANG) + ".html"
)
@app.route("/web/logout", methods=["GET"])
@ -55,7 +56,7 @@ def logout():
def admin_homepage():
if not (
"user" in session
and session["user"] == config.get(ConfigParameter.WEB_USERNAME)
and session["user"] == app.config["CONFIG"].get(ConfigParameter.WEB_USERNAME)
):
# TODO localization
flash("Vous avez été déconnecté.")
@ -63,9 +64,9 @@ def admin_homepage():
comments = dao.find_not_published_comments()
return render_template(
"admin_" + config.get(ConfigParameter.LANG) + ".html",
"admin_" + app.config["CONFIG"].get(ConfigParameter.LANG) + ".html",
comments=comments,
baseurl=config.get(ConfigParameter.SITE_URL),
baseurl=app.config["CONFIG"].get(ConfigParameter.SITE_URL),
)
@ -77,7 +78,7 @@ def admin_action():
flash("Commentaire introuvable")
elif request.form.get("action") == "APPROVE":
dao.publish_comment(comment)
rss.generate()
app.config["RSS"].generate()
# TODO localization
flash("Commentaire publié")
else:

93
src/stacosys/run.py
View file

@ -9,55 +9,36 @@ import sys
from stacosys.db import database
from stacosys.interface import api, app, form
from stacosys.interface.web import admin
from stacosys.service import config, mailer, rss
from stacosys.service.configuration import ConfigParameter
from stacosys.service.configuration import Config, ConfigParameter
from stacosys.service.mail import Mailer
from stacosys.service.rssfeed import Rss
# configure logging
def configure_logging(level):
root_logger = logging.getLogger()
root_logger.setLevel(level)
handler = logging.StreamHandler()
handler.setLevel(level)
formatter = logging.Formatter("[%(asctime)s] %(name)s %(levelname)s %(message)s")
handler.setFormatter(formatter)
root_logger.addHandler(handler)
def stacosys_server(config_pathname):
# configure logging
def configure_logging() -> logging.Logger:
logging.basicConfig(
level=logging.INFO, format="[%(asctime)s] %(name)s %(levelname)s %(message)s"
)
logger = logging.getLogger(__name__)
configure_logging(logging.INFO)
logging.getLogger("werkzeug").level = logging.WARNING
return logger
# check config file exists
def load_and_validate_config(config_pathname: str, logger: logging.Logger) -> Config:
if not os.path.isfile(config_pathname):
logger.error("Configuration file '%s' not found.", config_pathname)
sys.exit(1)
raise FileNotFoundError(f"Configuration file '{config_pathname}' not found.")
# load and check config
config = Config()
config.load(config_pathname)
is_config_ok, erreur_config = config.check()
if not is_config_ok:
logger.error("Configuration incorrecte '%s'", erreur_config)
sys.exit(1)
logger.info(config)
if not config.check():
raise ValueError(f"Invalid configuration '{config_pathname}'")
logger.info("Configuration loaded successfully.")
return config
# initialize database
database.configure(config.get(ConfigParameter.DB))
logger.info("Start Stacosys application")
# generate RSS
rss.configure(
config.get(ConfigParameter.RSS_FILE),
config.get(ConfigParameter.SITE_NAME),
config.get(ConfigParameter.SITE_PROTO),
config.get(ConfigParameter.SITE_URL),
)
rss.generate()
# configure mailer
def configure_and_validate_mailer(config, logger):
mailer = Mailer()
mailer.configure_smtp(
config.get(ConfigParameter.SMTP_HOST),
config.get_int(ConfigParameter.SMTP_PORT),
@ -65,11 +46,37 @@ def stacosys_server(config_pathname):
config.get(ConfigParameter.SMTP_PASSWORD),
)
mailer.configure_destination(config.get(ConfigParameter.SITE_ADMIN_EMAIL))
mailer.check()
if not mailer.check():
logger.error("Email configuration not working")
sys.exit(1)
return mailer
def configure_rss(config):
rss = Rss()
rss.configure(
config.get(ConfigParameter.RSS_FILE),
config.get(ConfigParameter.SITE_NAME),
config.get(ConfigParameter.SITE_PROTO),
config.get(ConfigParameter.SITE_URL),
)
rss.generate()
return rss
def main(config_pathname):
logger = configure_logging()
config = load_and_validate_config(config_pathname, logger)
database.configure(config.get(ConfigParameter.DB))
logger.info("Start Stacosys application")
rss = configure_rss(config)
mailer = configure_and_validate_mailer(config, logger)
logger.info("start interfaces %s %s %s", api, form, admin)
# start Flask
app.config["CONFIG"] = config
app.config["MAILER"] = mailer
app.config["RSS"] = rss
app.run(
host=config.get(ConfigParameter.HTTP_HOST),
port=config.get_int(ConfigParameter.HTTP_PORT),
@ -82,4 +89,8 @@ if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("config", help="config path name")
args = parser.parse_args()
stacosys_server(args.config)
try:
main(args.config)
except Exception as e:
logging.error(f"Failed to start application: {e}")
sys.exit(1)

10
src/stacosys/service/__init__.py
View file

@ -1,10 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from .configuration import Config
from .mail import Mailer
from .rssfeed import Rss
config = Config()
mailer = Mailer()
rss = Rss()

58
src/stacosys/service/mail.py
View file

@ -2,44 +2,41 @@
# -*- coding: utf-8 -*-
import logging
import smtplib
import ssl
from email.mime.text import MIMEText
from smtplib import SMTP_SSL, SMTPAuthenticationError
logger = logging.getLogger(__name__)
class Mailer:
def __init__(self) -> None:
self._smtp_host: str = ""
self._smtp_port: int = 0
self._smtp_login: str = ""
self._smtp_password: str = ""
self._site_admin_email: str = ""
self._smtp_host = ""
self._smtp_port = 0
self._smtp_login = ""
self._smtp_password = ""
self._site_admin_email = ""
def configure_smtp(
self,
smtp_host,
smtp_port,
smtp_login,
smtp_password,
self, smtp_host: str, smtp_port: int, smtp_login: str, smtp_password: str
) -> None:
self._smtp_host = smtp_host
self._smtp_port = smtp_port
self._smtp_login = smtp_login
self._smtp_password = smtp_password
def configure_destination(self, site_admin_email) -> None:
def configure_destination(self, site_admin_email: str) -> None:
self._site_admin_email = site_admin_email
def check(self):
server = smtplib.SMTP_SSL(
self._smtp_host, self._smtp_port, context=ssl.create_default_context()
)
server.login(self._smtp_login, self._smtp_password)
server.close()
def check(self) -> bool:
try:
with SMTP_SSL(self._smtp_host, self._smtp_port) as server:
server.login(self._smtp_login, self._smtp_password)
return True
except SMTPAuthenticationError:
logger.exception("Invalid credentials")
return False
def send(self, subject, message) -> bool:
def send(self, subject: str, message: str) -> bool:
sender = self._smtp_login
receivers = [self._site_admin_email]
@ -48,15 +45,14 @@ class Mailer:
msg["To"] = self._site_admin_email
msg["From"] = sender
# pylint: disable=bare-except
try:
server = smtplib.SMTP_SSL(
self._smtp_host, self._smtp_port, context=ssl.create_default_context()
)
server.login(self._smtp_login, self._smtp_password)
server.send_message(msg, sender, receivers)
server.close()
success = True
except:
success = False
return success
with SMTP_SSL(self._smtp_host, self._smtp_port) as server:
server.login(self._smtp_login, self._smtp_password)
server.send_message(msg, sender, receivers)
return True
except SMTPAuthenticationError:
logger.exception("Invalid credentials")
return False
except Exception as e:
logger.exception(f"Error sending email: {e}")
return False

5
tests/test_config.py
View file

@ -3,13 +3,14 @@
import pytest
from stacosys.service import config
from stacosys.service.configuration import ConfigParameter
from stacosys.service.configuration import Config, ConfigParameter
EXPECTED_DB = "sqlite://db.sqlite"
EXPECTED_HTTP_PORT = 8080
EXPECTED_LANG = "fr"
config = Config()
@pytest.fixture
def init_config():

6
tests/test_form.py
View file

@ -7,6 +7,9 @@ import pytest
from stacosys.db import database
from stacosys.interface import app, form
from stacosys.service.configuration import Config
from stacosys.service.mail import Mailer
from stacosys.service.rssfeed import Rss
@pytest.fixture
@ -14,6 +17,9 @@ def client():
logger = logging.getLogger(__name__)
database.configure("sqlite:memory://db.sqlite")
logger.info(f"start interface {form}")
app.config["CONFIG"] = Config()
app.config["MAILER"] = Mailer()
app.config["RSS"] = Rss()
return app.test_client()

3
tests/test_mail.py
View file

@ -3,10 +3,11 @@
import pytest
from stacosys.service import mailer
from stacosys.service.mail import Mailer
def test_configure_and_check():
mailer = Mailer()
mailer.configure_smtp("localhost", 2525, "admin", "admin")
mailer.configure_destination("admin@mydomain.com")
with pytest.raises(ConnectionRefusedError):

3
tests/test_rssfeed.py
View file

@ -1,8 +1,9 @@
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
from stacosys.service import rss
from stacosys.service.rssfeed import Rss
def test_configure():
rss = Rss()
rss.configure("comments.xml", "blog", "http", "blog.mydomain.com")