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

More pythonic singleton with module. Apply pylint recommandations

Browse source
This commit is contained in:
Yax 2022-11-25 19:57:37 +01:00
parent f231ed1cbb
commit b1c64d2cc8
14 changed files with 3557 additions and 684 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
Makefile
View file

@ -1,15 +1,15 @@
all: black test typehint lint
black:
isort --multi-line 3 --profile black stacosys/
black stacosys/
poetry run isort --multi-line 3 --profile black stacosys/
poetry run black stacosys/
test:
pytest
poetry run pytest
typehint:
mypy --ignore-missing-imports stacosys/
poetry run mypy --ignore-missing-imports stacosys/
lint:
pylint stacosys/
poetry run pylint stacosys/

4
config.ini
View file

@ -6,7 +6,8 @@ db_sqlite_file = db.sqlite
[site]
name = "My blog"
url = http://blog.mydomain.com
proto = https
url = https://blog.mydomain.com
admin_email = admin@mydomain.com
redirect = /redirect
@ -15,7 +16,6 @@ host = 127.0.0.1
port = 8100
[rss]
proto = https
file = comments.xml
[smtp]

3923
stacosys.sublime-workspace
View file

File diff suppressed because it is too large Load diff

66
stacosys/core/rss.py
View file

@ -1,66 +0,0 @@
#!/usr/bin/python
# -*- coding: UTF-8 -*-
from datetime import datetime
import markdown
import PyRSS2Gen
from stacosys.model.comment import Comment
class Rss:
def __init__(
self,
lang,
rss_file,
rss_proto,
site_name,
site_url,
):
self._lang = lang
self._rss_file = rss_file
self._rss_proto = rss_proto
self._site_name = site_name
self._site_url = site_url
def generate(self):
md = markdown.Markdown()
items = []
for row in (
Comment.select()
.where(Comment.published)
.order_by(-Comment.published)
.limit(10)
):
item_link = "%s://%s%s" % (
self._rss_proto,
self._site_url,
row.url,
)
items.append(
PyRSS2Gen.RSSItem(
title="%s - %s://%s%s"
% (
self._rss_proto,
row.author_name,
self._site_url,
row.url,
),
link=item_link,
description=md.convert(row.content),
guid=PyRSS2Gen.Guid("%s/%d" % (item_link, row.id)),
pubDate=row.published,
)
)
rss_title = 'Commentaires du site "%s"' % self._site_name
rss = PyRSS2Gen.RSS2(
title=rss_title,
link="%s://%s" % (self._rss_proto, self._site_url),
description=rss_title,
lastBuildDate=datetime.now(),
items=items,
)
rss.write_xml(open(self._rss_file, "w"), encoding="utf-8")

11
stacosys/interface/form.py
View file

@ -7,6 +7,8 @@ from flask import abort, redirect, request
from stacosys.db import dao
from stacosys.interface import app
from stacosys.service import config, mailer
from stacosys.service.configuration import ConfigParameter
logger = logging.getLogger(__name__)
@ -46,7 +48,7 @@ def new_form_comment():
# send notification e-mail asynchronously
submit_new_comment(comment)
return redirect(app.config.get("SITE_REDIRECT"), code=302)
return redirect(config.get(ConfigParameter.SITE_REDIRECT), code=302)
def check_form_data(posted_comment):
@ -57,7 +59,7 @@ def check_form_data(posted_comment):
@background.task
def submit_new_comment(comment):
site_url = app.config.get("SITE_URL")
site_url = config.get(ConfigParameter.SITE_URL)
comment_list = (
f"Web admin interface: {site_url}/web/admin",
"",
@ -72,8 +74,9 @@ def submit_new_comment(comment):
email_body = "\n".join(comment_list)
# send email to notify admin
subject = "STACOSYS " + app.config.get("SITE_NAME")
if app.config.get("MAILER").send(subject, email_body):
site_name = config.get(ConfigParameter.SITE_NAME)
subject = f"STACOSYS {site_name}"
if mailer.send(subject, email_body):
logger.debug("new comment processed")
# save notification datetime

19
stacosys/interface/web/admin.py
View file

@ -8,6 +8,8 @@ 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__)
@ -23,8 +25,8 @@ def index():
def is_login_ok(username, password):
hashed = hashlib.sha256(password.encode()).hexdigest().upper()
return (
app.config.get("WEB_USERNAME") == username
and app.config.get("WEB_PASSWORD") == hashed
config.get(ConfigParameter.WEB_USERNAME) == username
and config.get(ConfigParameter.WEB_PASSWORD) == hashed
)
@ -40,7 +42,7 @@ def login():
flash("Identifiant ou mot de passe incorrect")
return redirect("/web/login")
# GET
return render_template("login_" + app.config.get("LANG", "fr") + ".html")
return render_template("login_" + config.get(ConfigParameter.LANG) + ".html")
@app.route("/web/logout", methods=["GET"])
@ -51,16 +53,19 @@ def logout():
@app.route("/web/admin", methods=["GET"])
def admin_homepage():
if not ("user" in session and session["user"] == app.config.get("WEB_USERNAME")):
if not (
"user" in session
and session["user"] == config.get(ConfigParameter.WEB_USERNAME)
):
# TODO localization
flash("Vous avez été déconnecté.")
return redirect("/web/login")
comments = dao.find_not_published_comments()
return render_template(
"admin_" + app.config.get("LANG", "fr") + ".html",
"admin_" + config.get(ConfigParameter.LANG) + ".html",
comments=comments,
baseurl=app.config.get("SITE_URL"),
baseurl=config.get(ConfigParameter.SITE_URL),
)
@ -72,7 +77,7 @@ def admin_action():
flash("Commentaire introuvable")
elif request.form.get("action") == "APPROVE":
dao.publish_comment(comment)
app.config.get("RSS").generate()
rss.generate()
# TODO localization
flash("Commentaire publié")
else:

51
stacosys/run.py
View file

@ -6,12 +6,11 @@ import logging
import os
import sys
from stacosys.conf.config import Config, ConfigParameter
from stacosys.core.mailer import Mailer
from stacosys.core.rss import Rss
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
# configure logging
@ -37,16 +36,16 @@ def stacosys_server(config_pathname):
logger.error("Configuration file '%s' not found.", config_pathname)
sys.exit(1)
# load config
conf = Config.load(config_pathname)
is_config_ok, erreur_config = conf.check()
# load and check 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(conf)
logger.info(config)
# check database file exists (prevents from creating a fresh db)
db_pathname = conf.get(ConfigParameter.DB_SQLITE_FILE)
db_pathname = config.get(ConfigParameter.DB_SQLITE_FILE)
if not db_pathname or not os.path.isfile(db_pathname):
logger.error("Database file '%s' not found.", db_pathname)
sys.exit(1)
@ -57,39 +56,29 @@ def stacosys_server(config_pathname):
logger.info("Start Stacosys application")
# generate RSS
rss = Rss(
conf.get(ConfigParameter.LANG),
conf.get(ConfigParameter.RSS_FILE),
conf.get(ConfigParameter.RSS_PROTO),
conf.get(ConfigParameter.SITE_NAME),
conf.get(ConfigParameter.SITE_URL),
rss.configure(
config.get(ConfigParameter.RSS_FILE),
config.get(ConfigParameter.SITE_PROTO),
config.get(ConfigParameter.SITE_NAME),
config.get(ConfigParameter.SITE_URL),
)
rss.generate()
# configure mailer
mailer = Mailer(
conf.get(ConfigParameter.SMTP_HOST),
conf.get_int(ConfigParameter.SMTP_PORT),
conf.get(ConfigParameter.SMTP_LOGIN),
conf.get(ConfigParameter.SMTP_PASSWORD),
conf.get(ConfigParameter.SITE_ADMIN_EMAIL),
mailer.configure_smtp(
config.get(ConfigParameter.SMTP_HOST),
config.get_int(ConfigParameter.SMTP_PORT),
config.get(ConfigParameter.SMTP_LOGIN),
config.get(ConfigParameter.SMTP_PASSWORD),
)
mailer.configure_destination(config.get(ConfigParameter.SITE_ADMIN_EMAIL))
# inject config parameters into flask
app.config.update(LANG=conf.get(ConfigParameter.LANG))
app.config.update(SITE_NAME=conf.get(ConfigParameter.SITE_NAME))
app.config.update(SITE_URL=conf.get(ConfigParameter.SITE_URL))
app.config.update(SITE_REDIRECT=conf.get(ConfigParameter.SITE_REDIRECT))
app.config.update(WEB_USERNAME=conf.get(ConfigParameter.WEB_USERNAME))
app.config.update(WEB_PASSWORD=conf.get(ConfigParameter.WEB_PASSWORD))
app.config.update(MAILER=mailer)
app.config.update(RSS=rss)
logger.info("start interfaces %s %s %s", api, form, admin)
# start Flask
app.run(
host=conf.get(ConfigParameter.HTTP_HOST),
port=conf.get_int(ConfigParameter.HTTP_PORT),
host=config.get(ConfigParameter.HTTP_HOST),
port=config.get_int(ConfigParameter.HTTP_PORT),
debug=False,
use_reloader=False,
)

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

@ -0,0 +1,10 @@
#!/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()

32
stacosys/conf/config.py → stacosys/service/configuration.py
View file

@ -12,7 +12,6 @@ class ConfigParameter(Enum):
HTTP_HOST = "http.host"
HTTP_PORT = "http.port"
RSS_PROTO = "rss.proto"
RSS_FILE = "rss.file"
SMTP_HOST = "smtp.host"
@ -20,6 +19,7 @@ class ConfigParameter(Enum):
SMTP_LOGIN = "smtp.login"
SMTP_PASSWORD = "smtp.password"
SITE_PROTO = "site.proto"
SITE_NAME = "site.name"
SITE_URL = "site.url"
SITE_ADMIN_EMAIL = "site.admin_email"
@ -30,14 +30,16 @@ class ConfigParameter(Enum):
class Config:
def __init__(self):
self._cfg = configparser.ConfigParser()
@classmethod
def load(cls, config_pathname):
config = cls()
config._cfg.read(config_pathname)
return config
_cfg = configparser.ConfigParser()
# def __new__(cls):
# if not hasattr(cls, "instance"):
# cls.instance = super(Config, cls).__new__(cls)
# return cls.instance
def load(self, config_pathname):
self._cfg.read(config_pathname)
def _split_key(self, key: ConfigParameter):
section, param = str(key.value).split(".")
@ -50,12 +52,12 @@ class Config:
section, param = self._split_key(key)
return self._cfg.has_option(section, param)
def get(self, key: ConfigParameter):
def get(self, key: ConfigParameter) -> str:
section, param = self._split_key(key)
return (
self._cfg.get(section, param)
if self._cfg.has_option(section, param)
else None
else ""
)
def put(self, key: ConfigParameter, value):
@ -64,11 +66,11 @@ class Config:
self._cfg.add_section(section)
self._cfg.set(section, param, str(value))
def get_int(self, key: ConfigParameter):
def get_int(self, key: ConfigParameter) -> int:
value = self.get(key)
return int(value) if value else 0
def get_bool(self, key: ConfigParameter):
def get_bool(self, key: ConfigParameter) -> bool:
value = self.get(key)
assert value in (
"yes",
@ -85,8 +87,8 @@ class Config:
return (True, None)
def __repr__(self):
d = dict()
dict_repr = {}
for section in self._cfg.sections():
for option in self._cfg.options(section):
d[".".join([section, option])] = self._cfg.get(section, option)
return str(d)
dict_repr[".".join([section, option])] = self._cfg.get(section, option)
return str(dict_repr)

18
stacosys/core/mailer.py → stacosys/service/mail.py
View file

@ -10,21 +10,29 @@ logger = logging.getLogger(__name__)
class Mailer:
def __init__(
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 = ""
def configure_smtp(
self,
smtp_host,
smtp_port,
smtp_login,
smtp_password,
site_admin_email,
):
) -> 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:
self._site_admin_email = site_admin_email
def send(self, subject, message):
def send(self, subject, message) -> bool:
sender = self._smtp_login
receivers = [self._site_admin_email]
@ -34,7 +42,7 @@ class Mailer:
msg["From"] = sender
context = ssl.create_default_context()
# TODO catch SMTP failure
# TODO catch SMTP failure
with smtplib.SMTP_SSL(
self._smtp_host, self._smtp_port, context=context
) as server:

63
stacosys/service/rssfeed.py Normal file
View file

@ -0,0 +1,63 @@
#!/usr/bin/python
# -*- coding: UTF-8 -*-
from datetime import datetime
import markdown
import PyRSS2Gen
from stacosys.model.comment import Comment
class Rss:
def __init__(self) -> None:
self._rss_file: str = ""
self._site_proto: str = ""
self._site_name: str = ""
self._site_url: str = ""
def configure(
self,
rss_file,
site_proto,
site_name,
site_url,
) -> None:
self._rss_file = rss_file
self._site_proto = site_proto
self._site_name = site_name
self._site_url = site_url
def generate(self) -> None:
markdownizer = markdown.Markdown()
items = []
for row in (
Comment.select()
.where(Comment.published)
.order_by(-Comment.published)
.limit(10)
):
item_link = f"{self._site_proto}://{self._site_url}{row.url}"
items.append(
PyRSS2Gen.RSSItem(
title=f"{self._site_proto}://{self._site_url}{row.url} - {row.author_name}",
link=item_link,
description=markdownizer.convert(row.content),
guid=PyRSS2Gen.Guid(f"{item_link}{row.id}"),
pubDate=row.published,
)
)
rss_title = f"Commentaires du site {self._site_name}"
rss = PyRSS2Gen.RSS2(
title=rss_title,
link=f"{self._site_proto}://{self._site_url}",
description=rss_title,
lastBuildDate=datetime.now(),
items=items,
)
# TODO technical debt: replace pyRss2Gen
# TODO validate feed (https://validator.w3.org/feed/check.cgi)
# pylint: disable=consider-using-with
rss.write_xml(open(self._rss_file, "w", encoding="utf-8"), encoding="utf-8")

1
tests/test_api.py
View file

@ -25,7 +25,6 @@ def client():
logger = logging.getLogger(__name__)
database.setup(":memory:")
init_test_db()
app.config.update(SITE_TOKEN="ETC")
logger.info(f"start interface {api}")
return app.test_client()

30
tests/test_config.py
View file

@ -3,7 +3,8 @@
import unittest
from stacosys.conf.config import Config, ConfigParameter
from stacosys.service import config
from stacosys.service.configuration import ConfigParameter
EXPECTED_DB_SQLITE_FILE = "db.sqlite"
EXPECTED_HTTP_PORT = 8080
@ -11,31 +12,30 @@ EXPECTED_LANG = "fr"
class ConfigTestCase(unittest.TestCase):
def setUp(self):
self.conf = Config()
self.conf.put(ConfigParameter.DB_SQLITE_FILE, EXPECTED_DB_SQLITE_FILE)
self.conf.put(ConfigParameter.HTTP_PORT, EXPECTED_HTTP_PORT)
def setUp(self):
config.put(ConfigParameter.DB_SQLITE_FILE, EXPECTED_DB_SQLITE_FILE)
config.put(ConfigParameter.HTTP_PORT, EXPECTED_HTTP_PORT)
def test_exists(self):
self.assertTrue(self.conf.exists(ConfigParameter.DB_SQLITE_FILE))
self.assertTrue(config.exists(ConfigParameter.DB_SQLITE_FILE))
def test_get(self):
self.assertEqual(
self.conf.get(ConfigParameter.DB_SQLITE_FILE), EXPECTED_DB_SQLITE_FILE
config.get(ConfigParameter.DB_SQLITE_FILE), EXPECTED_DB_SQLITE_FILE
)
self.assertIsNone(self.conf.get(ConfigParameter.HTTP_HOST))
self.assertEqual(config.get(ConfigParameter.HTTP_HOST), "")
self.assertEqual(
self.conf.get(ConfigParameter.HTTP_PORT), str(EXPECTED_HTTP_PORT)
config.get(ConfigParameter.HTTP_PORT), str(EXPECTED_HTTP_PORT)
)
self.assertEqual(self.conf.get_int(ConfigParameter.HTTP_PORT), 8080)
self.assertEqual(config.get_int(ConfigParameter.HTTP_PORT), 8080)
try:
self.conf.get_bool(ConfigParameter.DB_SQLITE_FILE)
config.get_bool(ConfigParameter.DB_SQLITE_FILE)
self.assertTrue(False)
except AssertionError:
pass
def test_put(self):
self.assertFalse(self.conf.exists(ConfigParameter.LANG))
self.conf.put(ConfigParameter.LANG, EXPECTED_LANG)
self.assertTrue(self.conf.exists(ConfigParameter.LANG))
self.assertEqual(self.conf.get(ConfigParameter.LANG), EXPECTED_LANG)
self.assertFalse(config.exists(ConfigParameter.LANG))
config.put(ConfigParameter.LANG, EXPECTED_LANG)
self.assertTrue(config.exists(ConfigParameter.LANG))
self.assertEqual(config.get(ConfigParameter.LANG), EXPECTED_LANG)

3
tests/test_form.py
View file

@ -13,8 +13,7 @@ from stacosys.interface import form
@pytest.fixture
def client():
logger = logging.getLogger(__name__)
database.setup(":memory:")
app.config.update(SITE_REDIRECT="/redirect")
database.setup(":memory:")
logger.info(f"start interface {form}")
return app.test_client()