Antilint

mypy.ini

@@ -30,3 +30,6 @@ ignore_missing_imports = True
 
 [mypy-docker.*]
 ignore_missing_imports = True
+
+[mypy-mysql]
+ignore_missing_imports = True

scone/common/misc.py

@@ -16,8 +16,10 @@
 #  along with Scone.  If not, see <https://www.gnu.org/licenses/>.
 
 import os
+import re
 import sys
 from hashlib import sha256
+from typing import Dict
 
 
 def eprint(*args, **kwargs):
@@ -57,3 +59,25 @@ def sha256_file(path: str) -> str:
 
 def sha256_bytes(data: bytes) -> str:
     return sha256(data).hexdigest()
+
+
+def multireplace(string: str, replacements: Dict[str, str]) -> str:
+    """
+    Given a string and a replacement map, it returns the replaced string.
+
+    :param string: string to execute replacements on
+    :param replacements: replacement dictionary {value to find: value to replace}
+
+    source: https://stackoverflow.com/a/36620263
+    """
+    # Place longer ones first to keep shorter substrings from matching
+    # where the longer ones should take place
+    # For instance given the replacements {'ab': 'AB', 'abc': 'ABC'} against
+    # the string 'hey abc', it should produce 'hey ABC' and not 'hey ABc'
+    substrs = sorted(replacements, key=len, reverse=True)
+
+    # Create a big OR regex that matches any of the substrings to replace
+    regexp = re.compile("|".join(map(re.escape, substrs)))
+
+    # For each match, look up the new string in the replacements
+    return regexp.sub(lambda match: replacements[match.group(0)], string)

scone/default/recipes/apt.py

@@ -20,7 +20,7 @@ import logging
 import random
 from asyncio import Lock
 from collections import defaultdict
-from typing import List, Tuple, Dict
+from typing import Dict, List, Tuple
 
 from scone.default.utensils.basic_utensils import SimpleExec
 from scone.head.head import Head

scone/default/recipes/mysql.py

@@ -16,7 +16,7 @@
 #  along with Scone.  If not, see <https://www.gnu.org/licenses/>.
 from typing import List
 
-from scone.default.utensils.db_utensils import PostgresTransaction, MysqlTransaction
+from scone.default.utensils.db_utensils import MysqlTransaction
 from scone.head.head import Head
 from scone.head.kitchen import Kitchen, Preparation
 from scone.head.recipe import Recipe, RecipeContext
@@ -36,7 +36,11 @@ def mysql_dodgy_escape_username(unescaped: str) -> str:
     parts = unescaped.split("@")
     if len(parts) != 2:
         raise ValueError(f"{unescaped!r} is not a valid sconified mysql user name.")
-    return mysql_dodgy_escape_literal(parts[0]) + "@" + mysql_dodgy_escape_literal(parts[1])
+    return (
+        mysql_dodgy_escape_literal(parts[0])
+        + "@"
+        + mysql_dodgy_escape_literal(parts[1])
+    )
 
 
 class MysqlDatabase(Recipe):
@@ -59,12 +63,7 @@ class MysqlDatabase(Recipe):
 
     async def cook(self, kitchen: Kitchen) -> None:
         ch = await kitchen.start(MysqlTransaction("mysql", "root", unix_socket=True))
-        await ch.send(
-            (
-                "SHOW DATABASES LIKE %s",
-                self.database_name,
-            )
-        )
+        await ch.send(("SHOW DATABASES LIKE %s", self.database_name,))
         dbs = await ch.recv()
         if len(dbs) > 0:
             await ch.send(None)
@@ -93,7 +92,7 @@ class MysqlDatabase(Recipe):
                 if len(res) != 0:
                     raise RuntimeError("expected empty result set.")
 
-            q = f"""
+            q = """
                 FLUSH PRIVILEGES
             """
             await ch.send((q,))

scone/default/recipes/systemd.py

@@ -18,7 +18,8 @@
 from scone.default.steps.systemd_steps import (
     cook_systemd_daemon_reload,
     cook_systemd_enable,
-    cook_systemd_start, cook_systemd_stop,
+    cook_systemd_start,
+    cook_systemd_stop,
 )
 from scone.head.head import Head
 from scone.head.kitchen import Kitchen, Preparation

scone/default/utensils/db_utensils.py

@@ -123,7 +123,9 @@ class MysqlTransaction(Utensil):
 
         unix_socket = "/var/run/mysqld/mysqld.sock" if self.unix_socket else None
 
-        conn = mysql_connector.connect(database=self.database, user=self.user, unix_socket=unix_socket)
+        conn = mysql_connector.connect(
+            database=self.database, user=self.user, unix_socket=unix_socket
+        )
         cur = conn.cursor()
         try:
             await queryloop()

scone/sous/__main__.py

@@ -16,7 +16,6 @@
 #  along with Scone.  If not, see <https://www.gnu.org/licenses/>.
 
 import asyncio
-import itertools
 import logging
 import os
 import pwd
@@ -98,7 +97,9 @@ async def main(args: List[str]):
 
             logger.debug("going to sched task with %r", utensil)
 
-            awaitables.append(asyncio.create_task(run_utensil(utensil, channel, worktop)))
+            awaitables.append(
+                asyncio.create_task(run_utensil(utensil, channel, worktop))
+            )
         elif "lost" in message:
             # for a then-non-existent channel, but probably just waiting on us
             # retry without a default route.