Add argument parser and toml-based config file

- Create a common argument interface and config file format for both
  cli and flask version of application.
- Bumped the python version constraint to ^3.11 in order to use the
  native tomllib for reading config file.
- Move all the flask specific code under `worblehat.flaskapp`

config-template.toml

@@ -0,0 +1,20 @@
+# See https://flask.palletsprojects.com/en/2.3.x/config/
+[flask]
+TESTING = true
+DEBUG = true
+FLASK_ENV = 'development'
+SECRET_KEY = 'change-me'
+
+[database]
+# One of (sqlite, postgres)
+type = 'sqlite'
+
+[database.sqlite]
+path = './worblehat.sqlite'
+
+[database.postgres]
+host = 'localhost'
+port = 5432
+username = 'worblehat'
+password = 'change-me'
+name = 'worblehat'

pyproject.toml

@@ -13,7 +13,7 @@ flask = "^2.2.2"
 flask-admin = "^1.6.1"
 flask-sqlalchemy = "^3.0.3"
 isbnlib = "^3.10.14"
-python = "^3.10"
+python = "^3.11"
 python-dotenv = "^1.0.0"
 sqlalchemy = "^2.0.8"
 
@@ -22,7 +22,7 @@ werkzeug = "^2.3.3"
 
 [tool.poetry.scripts]
 cli = "worblehat.cli.main:main"
-dev = "worblehat.wsgi_dev:main"
+dev = "worblehat.flaskapp.wsgi_dev:main"
 
 [build-system]
 requires = ["poetry-core"]

worblehat/cli/main.py

@@ -8,17 +8,16 @@ from sqlalchemy import (
 from sqlalchemy.orm import (
    Session,
 )
-
 from worblehat.services.bookcase_item import (
     create_bookcase_item_from_isbn,
     is_valid_isbn,
 )
+from worblehat.services.config import Config
+from worblehat.services.argument_parser import parse_args
 
-from .prompt_utils import *
-
-from worblehat.config import Config
 from worblehat.models import *
 
+from .prompt_utils import *
 from .subclis.bookcase_item import BookcaseItemCli
 from .subclis.bookcase_shelf_selector import select_bookcase_shelf
 
@@ -30,14 +29,15 @@ class WorblehatCli(NumberedCmd):
     sql_session: Session
     sql_session_dirty: bool = False
 
-    def __init__(self):
+    def __init__(self, args: dict[str, any] | None = None):
         super().__init__()
 
         try:
-            engine = create_engine(Config.SQLALCHEMY_DATABASE_URI)
+            engine = create_engine(Config.db_string(), echo=args.get('verbose_sql', False))
             self.sql_session = Session(engine)
-        except Exception:
+        except Exception as err:
             print('Error: could not connect to database.')
+            print(err)
             exit(1)
 
         @event.listens_for(self.sql_session, 'after_flush')
@@ -51,7 +51,7 @@ class WorblehatCli(NumberedCmd):
             self.sql_session_dirty = False
             self.prompt_header = None
 
-        print(f"Debug: Connected to database at '{Config.SQLALCHEMY_DATABASE_URI}'")
+        print(f"Debug: Connected to database at '{Config.db_string()}'")
 
 
     def do_list_bookcases(self, _: str):
@@ -297,7 +297,10 @@ class WorblehatCli(NumberedCmd):
 
 
 def main():
-    tool = WorblehatCli()
+    args = parse_args()
+    Config.load_configuration(args)
+
+    tool = WorblehatCli(args)
     while True:
         try:
             tool.cmdloop()

worblehat/config.py

@@ -1,13 +0,0 @@
-from os import environ, path
-from dotenv import load_dotenv
-
-basedir = path.abspath(path.dirname(__file__))
-load_dotenv(path.join(basedir, '.env'))
-
-class Config:
-    TESTING = True
-    DEBUG = True
-    FLASK_ENV = 'development'
-    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + path.join(basedir, 'worblehat.sqlite')
-    SQLALCHEMY_TRACK_MODIFICATIONS = False
-    SECRET_KEY = environ.get('SECRET_KEY')

worblehat/flaskapp/flaskapp.py

@@ -3,23 +3,30 @@ from flask_admin import Admin
 from flask_admin.contrib.sqla import ModelView
 from sqlalchemy import inspect
 
-from .database import db
-from .models import *
-from .config import Config
+from worblehat.models import *
+from worblehat.services.seed_test_data import seed_data
+from worblehat.services.config import Config
 
 from .blueprints.main import main
-from .seed_test_data import seed_data
+from .database import db
 
-def create_app():
-    app = Flask(__name__, instance_relative_config=True)
-    app.config.from_object(Config)
+def create_app(args: dict[str, any] | None = None):
+    app = Flask(__name__)
+
+    if args is not None:
+        Config.load_configuration(args)
+        print(Config.db_string())
+        app.config.update(Config['flask'])
+        app.config.update(Config._config)
+        app.config['SQLALCHEMY_DATABASE_URI'] = Config.db_string()
+        app.config['SQLALCHEMY_ECHO'] = args.get('verbose_sql')
 
     db.init_app(app)
 
     with app.app_context():
         if not inspect(db.engine).has_table('Bookcase'):
             Base.metadata.create_all(db.engine)
-            seed_data(db)
+            seed_data()
 
     configure_admin(app)
 

worblehat/flaskapp/wsgi_dev.py

@@ -1,9 +1,13 @@
 from werkzeug import run_simple
 
+from worblehat.services.config import Config
+from worblehat.services.argument_parser import parse_args
+
 from .flaskapp import create_app
 
 def main():
-    app = create_app()
+    args = parse_args()
+    app = create_app(args)
     run_simple(
         hostname = 'localhost',
         port = 5000,

worblehat/models/mixins/UidMixin.py

@@ -3,28 +3,29 @@ from typing_extensions import Self
 from sqlalchemy import Integer
 from sqlalchemy.orm import (
     Mapped,
+    Session,
     mapped_column,
 )
 
-from worblehat.database import db
+from worblehat.flaskapp.database import db
 
 class UidMixin(object):
     uid: Mapped[int] = mapped_column(Integer, primary_key=True)
 
     @classmethod
-    def get_by_uid(cls, uid: int) -> Self | None:
+    def get_by_uid(cls, uid: int, sql_session: Session = db.session) -> Self | None:
         """
         NOTE:
-        This is a flask_sqlalchemy specific method.
-        It will not work outside of a request context.
+        This method defaults to using the flask_sqlalchemy session.
+        It will not work outside of a request context, unless another session is provided.
         """
-        return db.session.query(cls).where(cls.uid == uid).one_or_none()
+        return sql_session.query(cls).where(cls.uid == uid).one_or_none()
 
     @classmethod
-    def get_by_uid_or_404(cls, uid: int) -> Self:
+    def get_by_uid_or_404(cls, uid: int, sql_session: Session = db.session) -> Self:
         """
         NOTE:
-        This is a flask_sqlalchemy specific method.
-        It will not work outside of a request context.
+        This method defaults to using the flask_sqlalchemy session.
+        It will not work outside of a request context, unless another session is provided.
         """
-        return db.session.query(cls).where(cls.uid == uid).one_or_404()
+        return sql_session.query(cls).where(cls.uid == uid).one_or_404()

worblehat/models/mixins/UniqueNameMixin.py

@@ -3,28 +3,29 @@ from typing_extensions import Self
 from sqlalchemy import Text
 from sqlalchemy.orm import (
     Mapped,
+    Session,
     mapped_column,
 )
 
-from worblehat.database import db
+from worblehat.flaskapp.database import db
 
 class UniqueNameMixin(object):
     name: Mapped[str] = mapped_column(Text, unique=True, index=True)
 
     @classmethod
-    def get_by_name(cls, name: str) -> Self | None:
+    def get_by_name(cls, name: str, sql_session: Session = db.session) -> Self | None:
         """
         NOTE:
-        This is a flask_sqlalchemy specific method.
-        It will not work outside of a request context.
+        This method defaults to using the flask_sqlalchemy session.
+        It will not work outside of a request context, unless another session is provided.
         """
-        return db.session.query(cls).where(cls.name == name).one_or_none()
+        return sql_session.query(cls).where(cls.name == name).one_or_none()
 
     @classmethod
-    def get_by_uid_or_404(cls, name: str) -> Self:
+    def get_by_uid_or_404(cls, name: str, sql_session: Session = db.session) -> Self:
         """
         NOTE:
-        This is a flask_sqlalchemy specific method.
-        It will not work outside of a request context.
+        This method defaults to using the flask_sqlalchemy session.
+        It will not work outside of a request context, unless another session is provided.
         """
-        return db.session.query(cls).where(cls.name == name).one_or_404()
+        return sql_session.query(cls).where(cls.name == name).one_or_404()

worblehat/services/argument_parser.py

@@ -0,0 +1,62 @@
+from argparse import ArgumentParser
+from os import path
+from pathlib import Path
+from pprint import pformat
+
+def _print_version() -> None:
+    from worblehat import __version__
+    print(f'Worblehat version {__version__}')
+
+
+def _is_valid_file(parser: ArgumentParser, arg: str) -> Path:
+    path = Path(arg)
+    if not path.is_file():
+        parser.error(f'The file {arg} does not exist!')
+
+    return path
+
+
+def parse_args() -> dict[str, any]:
+    parser = ArgumentParser(
+        description = 'Worblehat library management system',
+    )
+
+    parser.add_argument(
+        '--verbose',
+        action = 'store_true',
+        help = 'Enable verbose mode',
+    )
+    parser.add_argument(
+        '--verbose-sql',
+        action = 'store_true',
+        help = 'Enable verbose SQL mode',
+    )
+    parser.add_argument(
+        '--version',
+        action = 'store_true',
+        help = 'Print version and exit',
+    )
+    parser.add_argument(
+        '--config',
+        type=lambda x: _is_valid_file(parser, x),
+        help = 'Path to config file',
+        dest = 'config_file',
+        metavar = 'FILE',
+    )
+    parser.add_argument(
+        '--print-config',
+        action = 'store_true',
+        help = 'Print configuration and quit',
+    )
+
+    args = parser.parse_args()
+
+    if args.version:
+        _print_version()
+        exit(0)
+
+    if args.print_config:
+        print(f'Configuration:\n{pformat(vars(args))}')
+        exit(0)
+
+    return vars(args)

worblehat/services/config.py

@@ -0,0 +1,74 @@
+from pathlib import Path
+import tomllib
+from typing import Any
+from pprint import pformat
+
+
+class Config:
+    _config = None
+    _expected_config_file_locations = [
+        Path('./config.toml'),
+        Path('~/.config/worblehat/config.toml'),
+        Path('/var/lib/worblehat/config.toml'),
+    ]
+
+    def __class_getitem__(cls, name: str) -> Any:
+        __config = cls._config
+        for attr in name.split('.'):
+            __config = __config.get(attr)
+            if __config is None:
+                raise AttributeError(f'No such attribute: {name}')
+        return __config
+
+
+    @classmethod
+    def _locate_configuration_file(cls) -> Path | None:
+        for path in cls._expected_config_file_locations:
+            if path.is_file():
+                return path
+
+
+    @classmethod
+    def _load_configuration_from_file(cls, config_file_path: str | None) -> dict[str, any]:
+        if config_file_path is None:
+            config_file_path = cls._locate_configuration_file()
+
+        if config_file_path is None:
+            print('Error: could not locate configuration file.')
+            exit(1)
+
+        with open(config_file_path, 'rb') as config_file:
+            args = tomllib.load(config_file)
+
+        return args
+
+
+    @classmethod
+    def db_string(cls) -> str:
+        db_type = cls._config.get('database').get('type')
+
+        if db_type == 'sqlite':
+            path = Path(cls._config.get('database').get('sqlite').get('path'))
+            return f"sqlite:///{path.absolute()}"
+
+        elif db_type == 'postgresql':
+            db_config = cls._config.get('database').get('postgresql')
+            hostname = db_config.get('hostname')
+            port = db_config.get('port')
+            username = db_config.get('username')
+            password = db_config.get('password')
+            database = db_config.get('database')
+            return f"psycopg2+postgresql://{username}:{password}@{hostname}:{port}/{database}"
+        else:
+            print(f"Error: unknown database type '{db_config.get('type')}'")
+            exit(1)
+
+
+    @classmethod
+    def debug(cls) -> str:
+        return pformat(cls._config)
+
+
+    @classmethod
+    def load_configuration(cls, args: dict[str, any]) -> dict[str, any]:
+        cls._config = cls._load_configuration_from_file(args.get('config_file'))

worblehat/services/seed_test_data.py

@@ -1,16 +1,18 @@
 import csv
 from pathlib import Path
 
-from flask_sqlalchemy import SQLAlchemy
+from sqlalchemy.orm import Session
 
-from .models import (
+from worblehat.flaskapp.database import db
+
+from ..models import (
     Bookcase,
     BookcaseShelf,
     Language,
     MediaType,
 )
 
-def seed_data(db: SQLAlchemy):
+def seed_data(sql_session: Session = db.session):
     media_types = [
         MediaType(name='Book', description='A physical book'),
         MediaType(name='Comic', description='A comic book'),
@@ -115,13 +117,13 @@ def seed_data(db: SQLAlchemy):
         BookcaseShelf(row=0, column=4, bookcase=bookcases[4], description="Religion"),
     ]
 
-    with open(Path(__file__).parent.parent / 'data' / 'iso639_1.csv') as f:
+    with open(Path(__file__).parent.parent.parent / 'data' / 'iso639_1.csv') as f:
       reader = csv.reader(f)
       languages = [Language(name, code) for (code, name) in reader]
 
-    db.session.add_all(media_types)
-    db.session.add_all(bookcases)
-    db.session.add_all(shelfs)
-    db.session.add_all(languages)
-    db.session.commit()
+    sql_session.add_all(media_types)
+    sql_session.add_all(bookcases)
+    sql_session.add_all(shelfs)
+    sql_session.add_all(languages)
+    sql_session.commit()
     print("Added test media types, bookcases and shelfs.")