remove unused functions

This commit is contained in:
ducklet 2023-03-28 22:04:14 +02:00
parent 84bbe331ee
commit e27b57050a

View file

@ -1,7 +1,6 @@
import asyncio import asyncio
import contextlib import contextlib
import logging import logging
import re
import threading import threading
from pathlib import Path from pathlib import Path
from typing import Any, Iterable, Literal, Type, TypeVar from typing import Any, Iterable, Literal, Type, TypeVar
@ -16,7 +15,6 @@ from .models import (
Rating, Rating,
User, User,
asplain, asplain,
fields,
fromplain, fromplain,
movies, movies,
optional_fields, optional_fields,
@ -414,7 +412,7 @@ async def add_or_update_rating(rating: Rating) -> bool:
return False return False
def sql_escape(s: str, char="#"): def sql_escape(s: str, char: str = "#") -> str:
return s.replace(char, 2 * char).replace("%", f"{char}%").replace("_", f"{char}_") return s.replace(char, 2 * char).replace("%", f"{char}%").replace("_", f"{char}_")
@ -533,10 +531,6 @@ async def ratings_for_movie_ids(
return tuple(dict(r._mapping) for r in rows) return tuple(dict(r._mapping) for r in rows)
def sql_fields(tp: Type):
return (f"{tp._table}.{f.name}" for f in fields(tp))
async def ratings_for_movies( async def ratings_for_movies(
movie_ids: Iterable[ULID], user_ids: Iterable[ULID] = [] movie_ids: Iterable[ULID], user_ids: Iterable[ULID] = []
) -> Iterable[Rating]: ) -> Iterable[Rating]:
@ -626,24 +620,3 @@ async def find_movies(
aggreg[rating.movie_id][1].append(rating) aggreg[rating.movie_id][1].append(rating)
return aggreg.values() return aggreg.values()
def bindparams(query: str, values: dict):
"""Bind values to a query.
This is similar to what SQLAlchemy and Databases do, but it allows to
easily use the same placeholder in multiple places.
"""
pump_vals = {}
pump_keys = {}
def pump(match):
key = match[1]
val = values[key]
pump_keys[key] = 1 + pump_keys.setdefault(key, 0)
pump_key = f"{key}_{pump_keys[key]}"
pump_vals[pump_key] = val
return f":{pump_key}"
pump_query = re.sub(r":(\w+)\b", pump, query)
return sa.text(pump_query).bindparams(**pump_vals)