datasette plugin hook to load "canned queries" from files in a sql subdirectory within the datasette "config_dir"

canned_queries_from_files.py

# Copyright 2021-2023 Mukunda Modell
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and 
# associated documentation files (the “Software”), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense,
# and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all copies or substantial
# portions of the Software.
#
# THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
# LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

from datasette.hookspecs import hookimpl
from datasette.utils import parse_metadata, BadMetadataError
from typing import Mapping
from .util import log

query_cache = {}

@hookimpl
def canned_queries(datasette, database: str) -> Mapping[str, str]:
    """ Load "canned queries" from .sql files stored under the config_dir.
    Looks for files named {config_dir}/sql/db/query_name.sql
    """
    queries = {}
    global query_cache
    if database is None:
        database = "*"

    if database in query_cache:
        return query_cache[database]

    sqldir = datasette.config_dir / "sql"
    if not sqldir.is_dir():
        query_cache[database] = queries
        return queries

    sqldirs = []
    sqldirs.append(sqldir)

    if database:
        sqldir = sqldir / database
        if sqldir.is_dir():
            sqldirs.append(sqldir)

    for sqldir in sqldirs:
        for f in sqldir.glob('*.sql'):
            try:
                sql = f.read_text('utf8').strip()
                if not len(sql):
                    log(f"Skipping empty canned query file: {f}")
                    continue

                try:
                    queries[f.stem] = parse_metadata(sql)
                except BadMetadataError as err:
                    queries[f.stem] = { "sql": sql }
            except OSError as err:
                log(err)

    query_cache[database] = queries
    return queries

util.py

class FakeConsole:
    """this is used if rich is not installed. This makes rich an optional dependency
     and keeps debugging code and console logging from breaking the world.
    """
    def log(self, *output):
        print(*output)

    def print_exception(self, e=None):
        print(e)

    def status(self, *msg):
        print(*msg)
        return self

    def update(self, *msg):
        print(*msg)
        return self

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        return False


try:
    from rich.console import Console
    console = Console(stderr=True)
except:
    console = FakeConsole()

   
def log(err):
    console.log(err)