Replace DB layer Peewee by PyDal

config.ini

@@ -2,7 +2,7 @@
 ; Default configuration
 [main]
 lang = fr
-db_sqlite_file = db.sqlite
+db = sqlite://db.sqlite
 
 [site]
 name = "My blog"

poetry.lock

@@ -315,14 +315,6 @@ category = "dev"
 optional = false
 python-versions = ">=3.7"
 
-[[package]]
-name = "peewee"
-version = "3.15.4"
-description = "a little orm"
-category = "main"
-optional = false
-python-versions = "*"
-
 [[package]]
 name = "platformdirs"
 version = "2.5.4"
@@ -355,6 +347,14 @@ category = "dev"
 optional = false
 python-versions = ">=3.6"
 
+[[package]]
+name = "pydal"
+version = "20221110.1"
+description = "a pure Python Database Abstraction Layer (for python version 2.7 and 3.x)"
+category = "main"
+optional = false
+python-versions = "*"
+
 [[package]]
 name = "pyflakes"
 version = "3.0.1"
@@ -524,7 +524,7 @@ python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,>=2.7"
 [metadata]
 lock-version = "1.1"
 python-versions = "~3.11"
-content-hash = "c7fd5e51d22b64ab20394250b12819609483e7e825996bb8c62e6bb2bb537951"
+content-hash = "eae5d8539c8fd2e80b005124c5439a61310128f7427bdc20affa92dec85085ee"
 
 [metadata.files]
 astroid = [
@@ -788,9 +788,6 @@ pathspec = [
     {file = "pathspec-0.10.2-py3-none-any.whl", hash = "sha256:88c2606f2c1e818b978540f73ecc908e13999c6c3a383daf3705652ae79807a5"},
     {file = "pathspec-0.10.2.tar.gz", hash = "sha256:8f6bf73e5758fd365ef5d58ce09ac7c27d2833a8d7da51712eac6e27e35141b0"},
 ]
-peewee = [
-    {file = "peewee-3.15.4.tar.gz", hash = "sha256:2581520c8dfbacd9d580c2719ae259f0637a9e46eda47dfc0ce01864c6366205"},
-]
 platformdirs = [
     {file = "platformdirs-2.5.4-py3-none-any.whl", hash = "sha256:af0276409f9a02373d540bf8480021a048711d572745aef4b7842dad245eba10"},
     {file = "platformdirs-2.5.4.tar.gz", hash = "sha256:1006647646d80f16130f052404c6b901e80ee4ed6bef6792e1f238a8969106f7"},
@@ -803,6 +800,9 @@ pycodestyle = [
     {file = "pycodestyle-2.10.0-py2.py3-none-any.whl", hash = "sha256:8a4eaf0d0495c7395bdab3589ac2db602797d76207242c17d470186815706610"},
     {file = "pycodestyle-2.10.0.tar.gz", hash = "sha256:347187bdb476329d98f695c213d7295a846d1152ff4fe9bacb8a9590b8ee7053"},
 ]
+pydal = [
+    {file = "pydal-20221110.1.tar.gz", hash = "sha256:7c3e891c70f8d8918e36276f210a1959bb7badf3b276f47191986ffcf5b6a390"},
+]
 pyflakes = [
     {file = "pyflakes-3.0.1-py2.py3-none-any.whl", hash = "sha256:ec55bf7fe21fff7f1ad2f7da62363d749e2a470500eab1b555334b67aa1ef8cf"},
     {file = "pyflakes-3.0.1.tar.gz", hash = "sha256:ec8b276a6b60bd80defed25add7e439881c19e64850afd9b346283d4165fd0fd"},

pyproject.toml

@@ -12,10 +12,10 @@ pyrss2gen = "^1.1"
 markdown = "^3.1.1"
 requests = "^2.25.1"
 coverage = "^6.5"
-peewee = "^3.14.8"
 background = "^0.2.1"
 Flask = "^2.1.1"
 types-markdown = "^3.4.2.1"
+pydal = "^20221110.1"
 
 [tool.poetry.group.dev.dependencies]
 pylint = "^2.15"

stacosys/db/__init__.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from pydal import DAL, Field
+
+
+class Database:
+
+    db_dal = DAL()
+
+    def configure(self, db_uri):
+        self.db_dal = DAL(db_uri, migrate=False)
+        self.db_dal.define_table(
+            "comment",
+            Field("url"),
+            Field("created", type="datetime"),
+            Field("notified", type="datetime"),
+            Field("published", type="datetime"),
+            Field("author_name"),
+            Field("author_site"),
+            Field("author_gravatar"),
+            Field("content", type="text"),
+        )
+
+    def get(self):
+        return self.db_dal
+
+
+database = Database()
+db = database.get

stacosys/db/dao.py

@@ -1,67 +1,80 @@
 #!/usr/bin/python
 # -*- coding: UTF-8 -*-
+# pylint: disable=singleton-comparison
+
 from datetime import datetime
 
+from stacosys.db import db
 from stacosys.model.comment import Comment
 
-TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
-
 
 def find_comment_by_id(comment_id):
-    return Comment.get_by_id(comment_id)
+    return db().comment(comment_id)
 
 
 def notify_comment(comment: Comment):
-    comment.notified = datetime.now().strftime(TIME_FORMAT)
+    db()(db().comment.id == comment.id).update(notified=datetime.now())
-    comment.save()
+    db().commit()
 
 
 def publish_comment(comment: Comment):
-    comment.published = datetime.now().strftime(TIME_FORMAT)
+    db()(db().comment.id == comment.id).update(published=datetime.now())
-    comment.save()
+    db().commit()
 
 
 def delete_comment(comment: Comment):
-    comment.delete_instance()
+    db()(db().comment.id == comment.id).delete()
+    db().commit()
 
 
 def find_not_notified_comments():
-    return Comment.select().where(Comment.notified.is_null())
+    return db()(db().comment.notified == None).select()
 
 
 def find_not_published_comments():
-    return Comment.select().where(Comment.published.is_null())
+    return db()(db().comment.published == None).select()
 
 
 def find_published_comments_by_url(url):
-    return (
+    return db()((db().comment.url == url) & (db().comment.published != None)).select(
-        Comment.select(Comment)
+        orderby=db().comment.published
-        .where((Comment.url == url) & (Comment.published.is_null(False)))
-        .order_by(+Comment.published)
     )
 
 
 def count_published_comments(url):
     return (
-        Comment.select(Comment)
+        db()((db().comment.url == url) & (db().comment.published != None)).count()
-        .where((Comment.url == url) & (Comment.published.is_null(False)))
-        .count()
         if url
-        else Comment.select(Comment).where(Comment.published.is_null(False)).count()
+        else db()(db().comment.published != None).count()
+    )
+
+
+def find_recent_published_comments():
+    return db()(db().comment.published != None).select(
+        orderby=~db().comment.published, limitby=(0, 10)
     )
 
 
 def create_comment(url, author_name, author_site, author_gravatar, message):
-    created = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    row = db().comment.insert(
-    comment = Comment(
         url=url,
         author_name=author_name,
         author_site=author_site,
         author_gravatar=author_gravatar,
         content=message,
-        created=created,
+        created=datetime.now(),
         notified=None,
         published=None,
     )
-    comment.save()
+    db().commit()
-    return comment
+    return Comment(
+        id=row.id,
+        url=row.url,
+        author_name=row.author_name,
+        author_site=row.author_site,
+        author_gravatar=row.author_gravatar,
+        content=row.content,
+        created=row.created,
+        notified=row.notified,
+        published=row.published,
+    )

stacosys/db/database.py

@@ -1,26 +0,0 @@
-#!/usr/bin/python
-# -*- coding: UTF-8 -*-
-# pylint: disable=import-outside-toplevel
-
-from peewee import Model
-from playhouse.db_url import SqliteDatabase
-
-db = SqliteDatabase(None)
-
-
-class BaseModel(Model):
-    class Meta:
-        database = db
-
-
-def setup(db_url):
-    db.init(db_url)
-    db.connect()
-
-    from stacosys.model.comment import Comment
-
-    db.create_tables([Comment], safe=True)
-
-
-def get_db():
-    return db

stacosys/model/comment.py

@@ -1,17 +1,19 @@
 #!/usr/bin/python
 # -*- coding: UTF-8 -*-
 
-from peewee import CharField, DateTimeField, TextField
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
 
-from stacosys.db.database import BaseModel
 
-
+@dataclass
-class Comment(BaseModel):
+class Comment:
-    url = CharField()
+    id: int = 0
-    created = DateTimeField()
+    url: str = ""
-    notified = DateTimeField(null=True, default=None)
+    created: Optional[datetime] = None
-    published = DateTimeField(null=True, default=None)
+    notified: Optional[datetime] = None
-    author_name = CharField()
+    published: Optional[datetime] = None
-    author_site = CharField(default="")
+    author_name: str = ""
-    author_gravatar = CharField(default="")
+    author_site: str = ""
-    content = TextField()
+    author_gravatar: str = ""
+    content: str = ""

stacosys/run.py

@@ -43,14 +43,8 @@ def stacosys_server(config_pathname):
         sys.exit(1)
     logger.info(config)
 
-    # check database file exists (prevents from creating a fresh db)
-    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)
-
     # initialize database
-    database.setup(db_pathname)
+    database.configure(config.get(ConfigParameter.DB))
 
     logger.info("Start Stacosys application")
 

stacosys/service/configuration.py

@@ -6,7 +6,7 @@ from enum import Enum
 
 
 class ConfigParameter(Enum):
-    DB_SQLITE_FILE = "main.db_sqlite_file"
+    DB = "main.db"
     LANG = "main.lang"
 
     HTTP_HOST = "http.host"

stacosys/service/rssfeed.py

@@ -6,7 +6,7 @@ from datetime import datetime
 import markdown
 import PyRSS2Gen
 
-from stacosys.model.comment import Comment
+from stacosys.db import dao
 
 
 class Rss:
@@ -32,12 +32,7 @@ class Rss:
         markdownizer = markdown.Markdown()
 
         items = []
-        for row in (
+        for row in dao.find_recent_published_comments():
-            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(

tests/test_api.py

@@ -23,7 +23,7 @@ def init_test_db():
 @pytest.fixture
 def client():
     logger = logging.getLogger(__name__)
-    database.setup(":memory:")
+    database.configure("sqlite:memory://db.sqlite")
     init_test_db()
     logger.info(f"start interface {api}")
     return app.test_client()

tests/test_config.py

@@ -6,26 +6,26 @@ import pytest
 from stacosys.service import config
 from stacosys.service.configuration import ConfigParameter
 
-EXPECTED_DB_SQLITE_FILE = "db.sqlite"
+EXPECTED_DB_SQLITE_FILE = "sqlite://db.sqlite"
 EXPECTED_HTTP_PORT = 8080
 EXPECTED_LANG = "fr"
 
 
 @pytest.fixture
 def init_config():
-    config.put(ConfigParameter.DB_SQLITE_FILE, EXPECTED_DB_SQLITE_FILE)
+    config.put(ConfigParameter.DB, EXPECTED_DB_SQLITE_FILE)
     config.put(ConfigParameter.HTTP_PORT, EXPECTED_HTTP_PORT)    
 
 def test_exists(init_config):
-    assert config.exists(ConfigParameter.DB_SQLITE_FILE)
+    assert config.exists(ConfigParameter.DB)
 
 def test_get(init_config):
-    assert config.get(ConfigParameter.DB_SQLITE_FILE) == EXPECTED_DB_SQLITE_FILE
+    assert config.get(ConfigParameter.DB) == EXPECTED_DB_SQLITE_FILE
     assert config.get(ConfigParameter.HTTP_HOST) == ""
     assert config.get(ConfigParameter.HTTP_PORT) == str(EXPECTED_HTTP_PORT)
     assert config.get_int(ConfigParameter.HTTP_PORT) == EXPECTED_HTTP_PORT
     with pytest.raises(AssertionError):
-        config.get_bool(ConfigParameter.DB_SQLITE_FILE)
+        config.get_bool(ConfigParameter.DB)
 
 def test_put(init_config):
     assert not config.exists(ConfigParameter.LANG)

tests/test_db.py

@@ -8,7 +8,7 @@ from stacosys.db import database
 
 @pytest.fixture
 def setup_db():
-    database.setup(":memory:")    
+    database.configure("sqlite:memory://db.sqlite")
 
 
 def test_dao_published(setup_db):

tests/test_form.py

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