Richie/dotfiles
1
0
Fork 0
Code Issues 1 Pull Requests 3 Actions 1 Packages Projects Releases Wiki Activity

replacing my python with per servis python setups #86

Merged
Richie merged 5 commits from feature/replacing-my_python-with-per-servis-python-setups into main 2026-09-19 21:11:37 -04:00
Conversation 0 Commits 5 Files Changed 20
+107 -504
20 changed files with 107 additions and 504 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/merge_flake_lock_update.yml
-1
View File
@@ -15,7 +15,6 @@ jobs:
steps:
- name: merge_flake_lock_update
run: >-
nix develop .#devShells.x86_64-linux.default -c
python -m python.gitea_flake_lock merge
--repo "${{ github.repository }}"
env:
.github/workflows/pytest.yml
-19
View File
@@ -1,19 +0,0 @@
name: pytest
on:
workflow_dispatch:
push:
branches:
- main
pull_request:
branches:
- main
jobs:
pytest:
runs-on: self-hosted
steps:
- uses: actions/checkout@v4
- name: Run tests
run: nix develop .#devShells.x86_64-linux.default -c pytest tests
.github/workflows/update-flake-lock.yml
-1
View File
@@ -21,6 +21,5 @@ jobs:
JEEVES_BOT_TOKEN: ${{ secrets.JEEVES_BOT_TOKEN }}
GITEA_URL: https://gitea.tmmworkshop.com
run: >-
nix develop .#devShells.x86_64-linux.default -c
python -m python.gitea_flake_lock update
--repo "${{ github.repository }}"
common/global/programs.nix
-1
View File
@@ -2,6 +2,5 @@
{
environment.systemPackages = with pkgs; [
git
my_python
];
}
common/optional/zfs/snapshot.nix
+42 -1
View File
@@ -1,4 +1,5 @@
{
inputs,
pkgs,
lib,
config,
@@ -6,6 +7,11 @@
}:
let
cfg = config.services.snapshot_manager;
snapshotManagerPackages =
ps: with ps; [
apprise
typer
];
in
{
options = {
@@ -34,6 +40,41 @@ in
};
config = lib.mkIf cfg.enable {
nixpkgs.overlays = [
(final: _prev: {
snapshot_manager_python = final.python314.withPackages snapshotManagerPackages;
snapshot_manager_test_python = final.python314.withPackages (
ps:
snapshotManagerPackages ps
++ (with ps; [
pyfakefs
pytest
pytest-asyncio
pytest-mock
pytest-xdist
])
);
snapshot_manager_tests =
final.runCommand "snapshot-manager-tests"
{
nativeBuildInputs = [ final.snapshot_manager_test_python ];
}
''
export HOME="$TMPDIR"
cd ${inputs.self}
pytest \
-o cache_dir="$TMPDIR/pytest-cache" \
tests/test_common.py \
tests/test_signal_alert.py \
tests/test_snapshot_manager.py \
tests/test_zfs.py
touch "$out"
'';
})
];
system.checks = [ pkgs.snapshot_manager_tests ];
systemd = {
services.snapshot_manager = {
description = "ZFS Snapshot Manager";
@@ -45,7 +86,7 @@ in
};
serviceConfig = {
Type = "oneshot";
ExecStart = "${pkgs.my_python}/bin/python -m python.tools.snapshot_manager ${lib.escapeShellArg cfg.path}";
ExecStart = "${pkgs.snapshot_manager_python}/bin/python -m python.tools.snapshot_manager ${lib.escapeShellArg cfg.path}";
}
// lib.optionalAttrs (cfg.EnvironmentFile != null) {
EnvironmentFile = cfg.EnvironmentFile;
overlays/default.nix
-31
View File
@@ -17,35 +17,4 @@
test-exclusions = import ./test-exclusions.nix;
x86-64-v3-workarounds = import ./x86-64-v3-workarounds.nix;
python-env = final: _prev: {
my_python = final.python314.withPackages (
ps: with ps; [
alembic
apprise
fastapi
fastapi-cli
httpx
jinja2
mypy
pgvector
psycopg
pydantic
pyfakefs
pytest
pytest-cov
pytest-mock
pytest-xdist
python-multipart
pydantic-settings
ruff
sqlalchemy
tenacity
tinytuya
typer
uvicorn
websockets
]
);
};
}
pyproject.toml
-43
View File
@@ -6,49 +6,6 @@ authors = [{ name = "Richie Cahill", email = "richie@tmmworkshop.com" }]
requires-python = "~=3.14.0"
readme = "README.md"
license = "MIT"
# these dependencies are a best effort and aren't guaranteed to work
# for up-to-date dependencies, see overlays/default.nix
dependencies = [
"alembic",
"apprise",
"beautifulsoup4",
"bm25s",
"ebooklib",
"fastapi",
"fastapi-cli",
"httpx",
"jinja2",
"pgvector",
"psycopg[binary]",
"pydantic",
"pydantic-settings",
"python-multipart",
"sqlalchemy[asyncio]",
"tenacity",
"tiktoken",
"tinytuya",
"typer",
"uvicorn",
"websockets",
"yake",
]
[project.scripts]
database = "python.database_cli:app"
whisper-transcribe = "python.tools.whisper.transcribe:main"
[dependency-groups]
dev = [
"aiosqlite",
"mypy",
"pyfakefs",
"pytest-asyncio",
"pytest-cov",
"pytest-mock",
"pytest-xdist",
"pytest",
"ruff",
]
[tool.ruff]
python/gitea_flake_lock.py
+1 -1
View File
@@ -14,7 +14,7 @@ DEFAULT_BASE_BRANCH = "main"
DEFAULT_BRANCH = "automation/update-flake-lock"
DEFAULT_GITEA_URL = "https://gitea.tmmworkshop.com"
PR_LABELS = ["dependencies", "automated", "flake_lock_update"]
PR_CHECK_WORKFLOWS = ["build_systems.yml", "treefmt.yml", "pytest.yml"]
PR_CHECK_WORKFLOWS = ["build_systems.yml", "treefmt.yml"]
PR_TITLE = "Update flake.lock"
PR_BODY = "Automated flake.lock update."
python/system_tests/__init__.py
-1
View File
@@ -1 +0,0 @@
"""system_tests."""
python/system_tests/components.py
-99
View File
@@ -1,99 +0,0 @@
"""Validate Jeeves."""
from __future__ import annotations
import logging
from copy import copy
from re import search
from time import sleep
from typing import TYPE_CHECKING
from python.common import bash_wrapper
from python.zfs import Zpool
if TYPE_CHECKING:
from collections.abc import Sequence
logger = logging.getLogger(__name__)
def zpool_tests(pool_names: Sequence[str], zpool_capacity_threshold: int = 90) -> list[str] | None:
"""Check the zpool health and capacity.
Args:
pool_names (Sequence[str]): A list of pool names to test.
zpool_capacity_threshold (int, optional): The threshold for the zpool capacity. Defaults to 90.
Returns:
list[str] | None: A list of errors if any.
"""
logger.info("Testing zpool")
errors: list[str] = []
for pool_name in pool_names:
pool = Zpool(pool_name)
if pool.health != "ONLINE":
errors.append(f"{pool.name} is {pool.health}")
if pool.capacity >= zpool_capacity_threshold:
errors.append(f"{pool.name} is low on space")
upgrade_status, _ = bash_wrapper("zpool upgrade")
if not search(r"Every feature flags pool has all supported and requested features enabled.", upgrade_status):
errors.append("ZPool out of date run `sudo zpool upgrade -a`")
return errors
def systemd_tests(
service_names: Sequence[str],
max_retries: int = 30,
retry_delay_secs: int = 1,
retryable_statuses: Sequence[str] | None = None,
valid_statuses: Sequence[str] | None = None,
) -> list[str] | None:
"""Tests a systemd services.
Args:
service_names (Sequence[str]): A list of service names to test.
max_retries (int, optional): The maximum number of retries. Defaults to 30.
minimum value is 1.
retry_delay_secs (int, optional): The delay between retries in seconds. Defaults to 1.
minimum value is 1.
retryable_statuses (Sequence[str] | None, optional): A list of retryable statuses. Defaults to None.
valid_statuses (Sequence[str] | None, optional): A list of valid statuses. Defaults to None.
Returns:
list[str] | None: A list of errors if any.
"""
logger.info("Testing systemd service")
max_retries = max(max_retries, 1)
retry_delay_secs = max(retry_delay_secs, 1)
last_try = max_retries - 1
if retryable_statuses is None:
retryable_statuses = ("inactive\n", "activating\n")
if valid_statuses is None:
valid_statuses = ("active\n",)
service_names_set = set(service_names)
errors: set[str] = set()
for retry in range(max_retries):
if not service_names_set:
break
logger.info(f"Testing systemd service in {retry + 1} of {max_retries}")
service_names_to_test = copy(service_names_set)
for service_name in service_names_to_test:
service_status, _ = bash_wrapper(f"systemctl is-active {service_name}")
if service_status in valid_statuses:
service_names_set.remove(service_name)
continue
if service_status in retryable_statuses and retry < last_try:
continue
errors.add(f"{service_name} is {service_status.strip()}")
sleep(retry_delay_secs)
return list(errors)
python/system_tests/validate_system.py
-67
View File
@@ -1,67 +0,0 @@
"""Validate {server_name}."""
import logging
import sys
import tomllib
from os import environ
from pathlib import Path # noqa: TC003 This is required for the typer CLI
from socket import gethostname
import typer
from python.common import configure_logger
from python.signal_alert import signal_alert
from python.system_tests.components import systemd_tests, zpool_tests
logger = logging.getLogger(__name__)
def load_config_data(config_file: Path) -> dict[str, list[str]]:
"""Load a TOML configuration file.
Args:
config_file (Path): The path to the configuration file.
Returns:
dict: The configuration data.
"""
return tomllib.loads(config_file.read_text())
def main(config_file: Path) -> None:
"""Main."""
configure_logger(level=environ.get("LOG_LEVEL", "INFO"))
server_name = gethostname()
logger.info(f"Starting {server_name} validation")
config_data = load_config_data(config_file)
errors: list[str] = []
try:
if config_data.get("zpools") and (zpool_errors := zpool_tests(config_data["zpools"])):
errors.extend(zpool_errors)
if config_data.get("services") and (systemd_errors := systemd_tests(config_data["services"])):
errors.extend(systemd_errors)
except Exception as error:
logger.exception(f"{server_name} validation failed")
errors.append(f"{server_name} validation failed: {error}")
if errors:
logger.error(f"{server_name} validation failed: \n{'\n'.join(errors)}")
signal_alert(f"{server_name} validation failed {errors}")
sys.exit(1)
logger.info(f"{server_name} validation passed")
def cli() -> None:
"""CLI."""
typer.run(main)
if __name__ == "__main__":
cli()
shell.nix
+1 -2
View File
@@ -8,7 +8,7 @@
# loader path, which NixOS does not provide globally.
ebook-search = pkgs.mkShell {
nativeBuildInputs = with pkgs; [
my_python
python314
uv
];
LD_LIBRARY_PATH = pkgs.lib.makeLibraryPath [
@@ -23,7 +23,6 @@
nix
home-manager
git
my_python
ssh-to-age
gnupg
systems/brain/services/heater.nix
+15 -1
View File
@@ -4,6 +4,20 @@
...
}:
{
nixpkgs.overlays = [
(final: _prev: {
heater_python = final.python314.withPackages (
ps: with ps; [
fastapi
pydantic
tinytuya
typer
uvicorn
]
);
})
];
networking.firewall.allowedTCPPorts = [ 8124 ];
systemd.services.heater-api = {
@@ -17,7 +31,7 @@
serviceConfig = {
Type = "simple";
ExecStart = "${pkgs.my_python}/bin/python -m python.heater.main --host 0.0.0.0 --port 8124";
ExecStart = "${pkgs.heater_python}/bin/python -m python.heater.main --host 0.0.0.0 --port 8124";
EnvironmentFile = "/etc/heater.env";
Restart = "on-failure";
RestartSec = "5s";
systems/jeeves/runners/nix_builder.nix
+8 -1
View File
@@ -23,6 +23,13 @@ let
...
}:
let
giteaAutomationPython = pkgs.python314.withPackages (
ps: with ps; [
httpx
pydantic
typer
]
);
runnerConfigFile = (pkgs.formats.yaml { }).generate "gitea-runner.yaml" { };
registerRunner = pkgs.writeShellApplication {
name = "register-gitea-runner";
@@ -96,8 +103,8 @@ let
curl
gawk
gitMinimal
giteaAutomationPython
gnused
my_python
nix
nixfmt
nixos-rebuild
systems/jeeves/services/gems.nix
+40 -1
View File
@@ -6,8 +6,47 @@
let
vars = import ../vars.nix;
stateDir = "${vars.services}/gems";
gemsPackages =
ps: with ps; [
fastapi
jinja2
pydantic
pydantic-settings
python-multipart
typer
uvicorn
];
in
{
nixpkgs.overlays = [
(final: _prev: {
gems_python = final.python314.withPackages gemsPackages;
gems_test_python = final.python314.withPackages (
ps:
gemsPackages ps
++ (with ps; [
httpx
pytest
pytest-asyncio
pytest-xdist
])
);
gems_tests =
final.runCommand "gems-tests"
{
nativeBuildInputs = [ final.gems_test_python ];
}
''
export HOME="$TMPDIR"
cd ${inputs.self}
pytest -o cache_dir="$TMPDIR/pytest-cache" tests/gems
touch "$out"
'';
})
];
system.checks = [ pkgs.gems_tests ];
users.groups.gems = { };
users.users.gems = {
isSystemUser = true;
@@ -36,7 +75,7 @@ in
Type = "simple";
User = "gems";
Group = "gems";
ExecStart = "${pkgs.my_python}/bin/python -m python.gems.main --host 0.0.0.0 --port 8002";
ExecStart = "${pkgs.gems_python}/bin/python -m python.gems.main --host 0.0.0.0 --port 8002";
Restart = "on-failure";
RestartSec = "5s";
StandardOutput = "journal";
systems/jeeves/services/systemd.nix
-26
View File
@@ -1,11 +1,7 @@
{
pkgs,
inputs,
...
}:
let
vars = import ../vars.nix;
in
{
systemd = {
services = {
@@ -30,21 +26,6 @@ in
ExecStart = "${pkgs.bash}/bin/bash -c 'echo 1 > /sys/bus/pci/devices/0000:61:00.0/remove'";
};
};
startup_validation = {
requires = [ "network-online.target" ];
after = [ "network-online.target" ];
wantedBy = [ "multi-user.target" ];
description = "validates startup";
path = [ pkgs.zfs ];
environment = {
PYTHONPATH = "${inputs.self}/";
};
serviceConfig = {
EnvironmentFile = "${vars.secrets}/services/server-validation";
Type = "oneshot";
ExecStart = "${pkgs.my_python}/bin/python -m python.system_tests.validate_system '${./validate_system.toml}'";
};
};
};
timers = {
plex_permission = {
@@ -55,13 +36,6 @@ in
Unit = "plex_permission.service";
};
};
startup_validation = {
wantedBy = [ "timers.target" ];
timerConfig = {
OnBootSec = "10min";
Unit = "startup_validation.service";
};
};
};
};
}
systems/jeeves/services/validate_system.toml
-6
View File
@@ -1,6 +0,0 @@
zpool = ["root_pool", "storage", "media"]
services = [
"audiobookshelf",
"docker",
"jellyfin",
]
systems/rhapsody-in-green/agent_logger.nix
-35
View File
@@ -1,35 +0,0 @@
{
pkgs,
inputs,
...
}:
{
systemd.services.agent-logger = {
description = "Unified agent logger";
after = [ "local-fs.target" ];
wantedBy = [ "multi-user.target" ];
environment = {
AGENT_LOG_DB = "/var/lib/agent-logger/agent_log.sqlite";
HOME = "/home/richie";
PYTHONPATH = "${inputs.self}";
};
serviceConfig = {
Type = "simple";
User = "richie";
WorkingDirectory = "/home/richie";
ExecStart = "${pkgs.my_python}/bin/python -m python.agent_logger.main";
StateDirectory = "agent-logger";
Restart = "on-failure";
RestartSec = "5s";
StandardOutput = "journal";
StandardError = "journal";
NoNewPrivileges = true;
ProtectSystem = "strict";
ProtectHome = "read-only";
PrivateTmp = true;
ReadOnlyPaths = [ "${inputs.self}" ];
};
};
}
tests/test_components.py
-104
View File
@@ -1,104 +0,0 @@
"""test_components."""
from pytest_mock import MockerFixture
from python.system_tests.components import systemd_tests, zpool_tests
from python.zfs import Zpool
temp = "Every feature flags pool has all supported and requested features enabled.\n"
SYSTEM_TESTS_COMPONENTS = "python.system_tests.components"
def test_zpool_tests(mocker: MockerFixture) -> None:
"""test_zpool_tests."""
mock_zpool = mocker.MagicMock(spec=Zpool)
mock_zpool.health = "ONLINE"
mock_zpool.capacity = 70
mock_zpool.name = "Main"
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.Zpool", return_value=mock_zpool)
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper", return_value=(temp, ""))
errors = zpool_tests(("Main",))
assert errors == []
def test_zpool_tests_out_of_date(mocker: MockerFixture) -> None:
"""test_zpool_tests_out_of_date."""
mock_zpool = mocker.MagicMock(spec=Zpool)
mock_zpool.health = "ONLINE"
mock_zpool.capacity = 70
mock_zpool.name = "Main"
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.Zpool", return_value=mock_zpool)
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper", return_value=("", ""))
errors = zpool_tests(("Main",))
assert errors == ["ZPool out of date run `sudo zpool upgrade -a`"]
def test_zpool_tests_out_of_space(mocker: MockerFixture) -> None:
"""test_zpool_tests_out_of_space."""
mock_zpool = mocker.MagicMock(spec=Zpool)
mock_zpool.health = "ONLINE"
mock_zpool.capacity = 100
mock_zpool.name = "Main"
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.Zpool", return_value=mock_zpool)
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper", return_value=(temp, ""))
errors = zpool_tests(("Main",))
assert errors == ["Main is low on space"]
def test_zpool_tests_offline(mocker: MockerFixture) -> None:
"""test_zpool_tests_offline."""
mock_zpool = mocker.MagicMock(spec=Zpool)
mock_zpool.health = "OFFLINE"
mock_zpool.capacity = 70
mock_zpool.name = "Main"
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.Zpool", return_value=mock_zpool)
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper", return_value=(temp, ""))
errors = zpool_tests(("Main",))
assert errors == ["Main is OFFLINE"]
def test_systemd_tests(mocker: MockerFixture) -> None:
"""test_systemd_tests."""
mocker.patch(
f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper",
side_effect=[
("inactive\n", ""),
("active\n", ""),
],
)
errors = systemd_tests(("docker",))
assert errors == []
"""test_systemd_tests."""
def test_systemd_tests_multiple_negative_retries(mocker: MockerFixture) -> None:
"""test_systemd_tests_fail."""
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper", return_value=("active\n", ""))
errors = systemd_tests(("docker",), max_retries=-1, retry_delay_secs=-1)
assert errors == []
def test_systemd_tests_multiple_pass(mocker: MockerFixture) -> None:
"""test_systemd_tests_fail."""
mocker.patch(
f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper",
side_effect=[
("inactive\n", ""),
("activating\n", ""),
("active\n", ""),
],
)
errors = systemd_tests(
("docker",),
retryable_statuses=("inactive\n", "activating\n"),
valid_statuses=("active\n",),
)
assert errors == []
def test_systemd_tests_fail(mocker: MockerFixture) -> None:
"""test_systemd_tests_fail."""
mocker.patch(f"{SYSTEM_TESTS_COMPONENTS}.bash_wrapper", return_value=("inactive\n", ""))
errors = systemd_tests(("docker",), max_retries=5)
assert errors == ["docker is inactive"]
tests/test_server_validate_scripts.py
-63
View File
@@ -1,63 +0,0 @@
"""test_server_validate_scripts."""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
from pytest_mock import MockerFixture
from python.system_tests.validate_system import main
if TYPE_CHECKING:
from pyfakefs.fake_filesystem import FakeFilesystem
from pytest_mock import MockerFixture
VALIDATE_SYSTEM = "python.system_tests.validate_system"
def test_validate_system(mocker: MockerFixture, fs: FakeFilesystem) -> None:
"""test_validate_system."""
fs.create_file(
"/mock_snapshot_config.toml",
contents='zpools = ["root_pool", "storage", "media"]\nservices = ["docker"]\n',
)
mocker.patch(f"{VALIDATE_SYSTEM}.systemd_tests", return_value=None)
mocker.patch(f"{VALIDATE_SYSTEM}.zpool_tests", return_value=None)
main(Path("/mock_snapshot_config.toml"))
def test_validate_system_errors(mocker: MockerFixture, fs: FakeFilesystem) -> None:
"""test_validate_system_errors."""
fs.create_file(
"/mock_snapshot_config.toml",
contents='zpools = ["root_pool", "storage", "media"]\nservices = ["docker"]\n',
)
mocker.patch(f"{VALIDATE_SYSTEM}.signal_alert")
mocker.patch(f"{VALIDATE_SYSTEM}.systemd_tests", return_value=["systemd_tests error"])
mocker.patch(f"{VALIDATE_SYSTEM}.zpool_tests", return_value=["zpool_tests error"])
with pytest.raises(SystemExit) as exception_info:
main(Path("/mock_snapshot_config.toml"))
assert exception_info.value.code == 1
def test_validate_system_execution(mocker: MockerFixture, fs: FakeFilesystem) -> None:
"""test_validate_system_execution."""
fs.create_file(
"/mock_snapshot_config.toml",
contents='zpools = ["root_pool", "storage", "media"]\nservices = ["docker"]\n',
)
mocker.patch(f"{VALIDATE_SYSTEM}.signal_alert")
mocker.patch(f"{VALIDATE_SYSTEM}.systemd_tests", return_value=None)
mocker.patch(f"{VALIDATE_SYSTEM}.zpool_tests", side_effect=RuntimeError("zpool_tests error"))
with pytest.raises(SystemExit) as exception_info:
main(Path("/mock_snapshot_config.toml"))
assert exception_info.value.code == 1