Coding

poetry init -n
poetry add python-fasthtml fastsql

cat << 'EOF' > main.py
from fasthtml.common import *
from fastsql import *
from sqlite_minutils.db import NotFoundError

app,rt,todos,Todo = fast_app(
    'data/todos.db',
    id=int, title=str, pk='id')

def tid(id): return f'todo-{id}'


@app.delete("/delete_todo", name='delete_todo')
async def delete_todo(id:int): 
    try: todos.delete(id)
    except NotFoundError: pass # If someone else deleted it already we don't have to do anything

@patch
def __ft__(self:Todo):
    show = Strong(self.title, target_id='current-todo')
    delete = A('delete',
               hx_delete=delete_todo.to(id=self.id).lstrip('/'), 
               hx_target=f'#{tid(self.id)}',
               hx_swap='outerHTML')
    return Li(show, ' | ', delete, id=tid(self.id))

def mk_input(**kw): return Input(id="new-title", name="title", placeholder="New Todo", **kw)

@rt
async def index():
    add =  Form(Group(mk_input(), Button("Add")), 
                post="insert_todo", target_id='todo-list', hx_swap="beforeend")
    card = Card(Ul(*todos(), id='todo-list'), header=add, footer=Div(id='current-todo')),
    title = 'Todo list'
    return Title(title), Main(H1(title), card, cls='container')

@rt
async def insert_todo(todo:Todo): return todos.insert(todo), mk_input(hx_swap_oob='true')

serve()
EOF

poetry run python main.py

cat << 'EOF' > main.py
#!/usr/bin/env python3

# Sample from https://blog.stoege.net/posts/python_decorator/

# Vars
a = 3
b = 8

# Decorator which logs start and end of a function
def log_function_call(func):
    def wrapper(*args, **kwargs):
        print(f"\nCalling function '{func.__name__}' with {args=} & {kwargs=}")
        result = func(*args, **kwargs)
        print(f"Function '{func.__name__}' finnished\n")
        return result
    return wrapper

@log_function_call
def add(a=int, b=int) -> int:
    result = a + b
    return result

@log_function_call
def sub(a=int, b=int) -> int:
    result = a - b
    return result

@log_function_call
def prod(a=int, b=int) -> int:
    result = a * b
    return result

@log_function_call
def div(a=int, b=int) -> int:
    result = a / b
    return result

if __name__ == "__main__":
    print("Result:", add(a, b))
    print("Result:", sub(a, b))
    print("Result:", prod(a, b))
    print("Result:", div(a, b))

EOF
chmod u+x main.py

##### RUN #####

# ./main.py

###############

user@host% ./main.py 

Calling function 'add' with args=(3, 8) & kwargs={}
Function 'add' finnished

Result: 11

Calling function 'sub' with args=(3, 8) & kwargs={}
Function 'sub' finnished

Result: -5

Calling function 'prod' with args=(3, 8) & kwargs={}
Function 'prod' finnished

Result: 24

Calling function 'div' with args=(3, 8) & kwargs={}
Function 'div' finnished

Result: 0.375

poetry init
poetry add motor pydantic
poetry add mkdocs-material mkdocstrings[python] mypy ruff poethepoet pytest pytest-asyncio pytest-random-order --group dev

[project]
name = "motorhead"
description = "Async MongoDB with vanilla Pydantic v2+ - made easy."
readme = "README.md"
license = { text = "MIT" }
authors = [
    { name = "Peter Volf", email = "[email protected]" },
]
requires-python = ">=3.10"
dependencies = ["pydantic", "motor"]
classifiers = [
    "Intended Audience :: Information Technology",
    "Operating System :: OS Independent",
    "Programming Language :: Python :: 3",
    "Development Status :: 4 - Beta",
    "Topic :: Internet",
    "Topic :: Software Development :: Libraries",
    "Topic :: Software Development",
    "Typing :: Typed",
    "Environment :: Web Environment",
    "Framework :: FastAPI",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: MIT License",
    "Topic :: Internet :: WWW/HTTP",
]

[project.urls]
homepage = "https://github.com/volfpeter/motorhead"
documentation = "https://volfpeter.github.io/motorhead"
tracker = "https://github.com/volfpeter/motorhead/issues"

[tool.poetry]
name = "motorhead"
version = "0.2403.0"
description = "Async MongoDB with vanilla Pydantic v2+ - made easy."
authors = ["Peter Volf <[email protected]>"]
readme = "README.md"
packages = [{include = "motorhead"}]

[tool.poetry.dependencies]
python = "^3.10"
motor = "^3.1.0"
pydantic = "^2.1.0"

[tool.poetry.group.dev.dependencies]
mkdocs-material = "^9.5.19"
mkdocstrings = {extras = ["python"], version = "^0.25.0"}
mypy = "^1.10.0"
ruff = "^0.4.2"
poethepoet = "^0.26.0"
pytest = "^8.2.0"
pytest-asyncio = "^0.23.6"
pytest-docker = "^3.1.1"
pytest-random-order = "^1.1.1"

[tool.mypy]
strict = true
show_error_codes = true
exclude = ["tree_app"]

[[tool.mypy.overrides]]
module = ["motor.*"]
ignore_missing_imports = true

[tool.ruff]
line-length = 108
lint.exclude = [
    ".git",
    ".mypy_cache",
    ".pytest_cache",
    ".ruff_cache",
    ".venv",
    "dist",
    "docs",
]
lint.select = [
    "E",  # pycodestyle errors
    "W",  # pycodestyle warnings
    "F",  # pyflakes
    "I",  # isort
    "S",  # flake8-bandit - we must ignore these rules in tests
    "C",  # flake8-comprehensions
    "B",  # flake8-bugbear
]

[tool.ruff.lint.per-file-ignores]
"tests/**/*" = ["S101"]  # S101: use of assert detected

[tool.pytest.ini_options]
addopts = "--random-order"

[tool.poe.tasks]
serve-docs = "mkdocs serve"
check-format = "ruff format --check ."
lint = "ruff check ."
mypy = "mypy ."
format = "ruff format ."
lint-fix = "ruff . --fix"
test = "python -m pytest tests --random-order"

static-checks.sequence = ["lint", "check-format", "mypy"]
static-checks.ignore_fail = "return_non_zero"

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

cat << 'EOF' > main.py
from twisted.web import server, resource
from twisted.internet import reactor, endpoints

class Counter(resource.Resource):
    isLeaf = True
    numberRequests = 0

    def render_GET(self, request):

        client_ip = request.getClientAddress().host

        r=request.uri.decode('utf-8')
        if not r =="/favicon.ico":
          self.numberRequests += 1

        request.setHeader(b"content-type", b"text/plain")
        content = u"I am request #{} from {}\n".format(self.numberRequests, client_ip)
        return content.encode("ascii")

endpoints.serverFromString(reactor, "tcp:8080").listen(server.Site(Counter()))
reactor.run()
EOF

poetry init
poetry add twisted
poetry run python main.py

cat <<'EOF'> ping.py
import argparse
from ping3 import ping, verbose_ping

def do_ping(host: str, timeout: int = 3, size: int = 1500, output: str = "json"):
    # output: json|txt
    # '21.54 ms'
    if size > 1500:
        size = 1500
    result = (
        str(
            round(
                ping(dest_addr=host, timeout=timeout, size=size, unit="ms"),
                2,
            )
        )
        + " ms"
    )
    if output.lower() == "json":
        return {"host": host, "timeout": timeout, "size": size, "result": result}
    if output.lower() == "txt":
        return result
    else:
        return f"output format '{output} unknown! use 'json|txt'"


def do_multiple_ping(host: str, count: int = 3, interval: float = 0):
    # ping 'www.stoege.net' ... 23ms
    # ping 'www.stoege.net' ... 24ms
    # ping 'www.stoege.net' ... 20ms
    verbose_ping(
        dest_addr=host,
        count=count,
        interval=interval,
    )


def main():
    # Create the argument parser
    parser = argparse.ArgumentParser(description="Ping a domain or IP address.")

    # Add the host argument
    parser.add_argument(
        "host",
        metavar="HOST",
        type=str,
        nargs="?",
        default="www.stoege.net",
        help="the domain or IP address to ping",
    )

    # Parse the command-line arguments
    args = parser.parse_args()

    # Call the ping function
    output = do_ping(host=args.host, output="json")

    # Print the ping output
    print(f"\n{output}\n")

    # Call the ping function. No return Value !
    do_multiple_ping(host=args.host, count=10, interval=0.1)


if __name__ == "__main__":
    main()
EOF

cat <<'EOF'> config.py
LOGGER_MAX_FILE_LENGTH = 10
EOF

mkdir src
cat <<'EOF'> src/logger.py
import logging
import datetime
import sys

from config import *

if isinstance(LOGGER_MAX_FILE_LENGTH, int):
    LOGGER_MAX_FILE_LENGTH = str(LOGGER_MAX_FILE_LENGTH)


def get_now() -> str:
    #
    # choose your format
    #
    current_time = datetime.datetime.now()

    # 2023-07-16 22:16:15.958
    formatted_time_1 = current_time.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]

    # 22:16:54.471
    formatted_time_2 = current_time.strftime("%H:%M:%S.%f")[:-3]

    # 22:17:21
    formatted_time_3 = current_time.strftime("%H:%M:%S")

    return formatted_time_2


class ExitOnCriticalHandler(logging.Handler):
    def emit(self, record):
        # Unset Color
        COLOR = RESET = "\033[0m"

        # Debug -> Black
        if record.levelno == logging.DEBUG:
            COLOR = "\033[0m"
        # Info -> Green
        elif record.levelno == logging.INFO:
            COLOR = "\033[92m"
        # Warn -> Blue
        elif record.levelno == logging.WARNING:
            COLOR = "\033[94m"
        # Error -> Orange
        elif record.levelno == logging.ERROR:
            COLOR = "\033[38;5;208m"
        # Critical -> Red
        elif record.levelno >= logging.CRITICAL:
            COLOR = "\033[91m"

        # Custom Line
        print(
            COLOR
            + "{:} {:} {:04} {:{w}} {:}".format(
                get_now(),
                record.levelname,
                record.lineno,
                record.filename,
                record.msg,
                w=LOGGER_MAX_FILE_LENGTH,
            )
            + RESET
        )

        # Exit on Critical
        if record.levelno >= logging.CRITICAL:
            logging.shutdown()
            print("\033[91m" + "GOT A CRITICAL -> EXIT HERE!" + "\033[0m")
            sys.exit()


# Init Logger
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Create and add the custom handler
exit_handler = ExitOnCriticalHandler()
logger.addHandler(exit_handler)


# Set Log Level
def set_log_level(level: str):
    # Parse/ Set LogLevel
    if level.lower() in ["d", "debug"]:
        logger.setLevel(logging.DEBUG)
    elif level.lower() in ["i", "info"]:
        logger.setLevel(logging.INFO)
    elif level.lower() in ["w", "warning"]:
        logger.setLevel(logging.WARNING)
    elif level.lower() in ["e", "error"]:
        logger.setLevel(logging.ERROR)
    elif level.lower() in ["c", "critical"]:
        logger.setLevel(logging.CRITICAL)

    # Custom level name mappings, Debug -> D
    logging.addLevelName(logging.DEBUG, "D")
    logging.addLevelName(logging.INFO, "I")
    logging.addLevelName(logging.WARNING, "W")
    logging.addLevelName(logging.ERROR, "E")
    logging.addLevelName(logging.CRITICAL, "C")


# Functions to Call
def ldebug(msg: str):
    logger.debug(msg)


def linfo(msg: str):
    logger.info(msg)


def lwarning(msg: str):
    logger.warning(msg)


def lerror(msg: str):
    logger.error(msg)


def lcritical(msg: str):
    logger.critical(msg)
EOF

export app="app100"
export FLASK_APP="${app}/app"
poetry new ${app}
cd ${app}

poetry env use $(which python3.10)
gsed -i "s/python = \"^3.*$/python = \"^3.10\"/" pyproject.toml
poetry lock

wget -4 -O requirements.txt https://raw.githubusercontent.com/GrahamMorbyDev/jwt-flask/master/requirements.txt
echo "marshmallow-sqlalchemy" >> requirements.txt
poetry add $(awk -F '==' '!/sha256/{print $1}' requirements.txt |tr '\n' ' ')
wget -4 -O ${app}/app.py https://raw.githubusercontent.com/GrahamMorbyDev/jwt-flask/master/app.py
poetry shell