Msc. lint stomping

This commit is contained in:
Reid 'arrdem' McKenzie 2021-08-30 00:30:44 -06:00
parent bf59df5ad7
commit 3612d2b9ee
6 changed files with 29 additions and 29 deletions

View file

@ -127,9 +127,9 @@ def available_migrations(queries: Queries, conn) -> t.Iterable[MigrationDescript
if query_name.endswith("_cursor"):
continue
# type: query_name: str
# query_name: str
# query_fn: t.Callable + {.__name__, .__doc__, .sql}
query_fn = getattr(queries, query_name)
# type: query_fn: t.Callable + {.__name__, .__doc__, .sql}
yield MigrationDescriptor(
name = query_name,
committed_at = None,

View file

@ -1,12 +1,12 @@
"""A quick and dirty Python driver for the jobqd API."""
from datetime import datetime
from typing import NamedTuple
import typing as t
import requests
class Job(NamedTuple):
class Job(t.NamedTuple):
id: int
payload: object
events: object
@ -39,7 +39,7 @@ class JobqClient(object):
.get("jobs"):
yield Job.from_json(job)
def poll(self, query, state) -> DehydratedJob:
def poll(self, query, state) -> Job:
"""Poll the job queue for the first job matching the given query, atomically advancing it to the given state and returning the advanced Job."""
return Job.from_json(
@ -49,7 +49,7 @@ class JobqClient(object):
"state": state})
.json())
def create(self, payload: object, state=None) -> DehydratedJob:
def create(self, payload: object, state=None) -> Job:
"""Create a new job in the system."""
return Job.from_json(
@ -59,7 +59,7 @@ class JobqClient(object):
"state": state})
.json())
def fetch(self, job: Job) -> HydratedJob:
def fetch(self, job: Job) -> Job:
"""Fetch the current state of a job."""
return Job.from_json(
@ -77,7 +77,7 @@ class JobqClient(object):
"new": state})
.json())
def event(self, job: Job, event: object) -> HydratedJob:
def event(self, job: Job, event: object) -> Job:
"""Attempt to record an event against a job."""
return Job.from_json(

View file

@ -54,14 +54,14 @@ def repl(opts, args, runtime):
try:
expr = parse_expr(line)
except Exception as e:
except Exception:
traceback.print_exc()
continue
try:
result = lil_eval(runtime, module, Bindings("__root__", None), expr)
print_([("class:result", f"{result!r}")], style=STYLE)
except Exception as e:
except Exception:
traceback.print_exc()
continue
@ -145,15 +145,15 @@ def main():
def py(runtime=None, module=None, expr=None, body=None, name=None):
"""The implementation of the Python lang as an eval type."""
g = globals().copy()
l = {}
globs = globals().copy()
locs = {}
body = (
f"def _shim():\n"
+ "\n".join(" " + l for l in body.splitlines())
"def _shim():\n"
+ "\n".join(" " + line for line in body.splitlines())
+ "\n\n_escape = _shim()"
)
exec(body, g, l)
return l["_escape"]
exec(body, globs, locs)
return locs["_escape"]
def lil(runtime=None, module=None, expr=None, body=None, name=None):
"""The implementation of the Lilith lang as an eval type."""

View file

@ -141,7 +141,7 @@ class Proquint(object):
"""Encode an integer into a proquint string."""
if width % 8 != 0 or width < 8:
raise ValueError(f"Width must be a positive power of 2 greater than 8")
raise ValueError("Width must be a positive power of 2 greater than 8")
return cls.encode_bytes(val.to_bytes(width // 8, "big"))