Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions wp1/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@ class AssessmentKind(enum.Enum):
TS_FORMAT_WP10 = "%Y%m%d%H%M%S"
LOG_DATE_FORMAT = "%B %-d, %Y"

GLOBAL_TIMESTAMP = time.strftime(TS_FORMAT_WP10, time.gmtime()).encode("utf-8")
GLOBAL_TIMESTAMP_WIKI = time.strftime(TS_FORMAT, time.gmtime()).encode("utf-8")
GLOBAL_TIMESTAMP: bytes = time.strftime(TS_FORMAT_WP10, time.gmtime()).encode("utf-8")
GLOBAL_TIMESTAMP_WIKI: bytes = time.strftime(TS_FORMAT, time.gmtime()).encode("utf-8")

LIST_URL = "https://tools.wmflabs.org/enwp10/cgi-bin/list2.fcgi"
LIST_V2_URL = "https://wp1.openzim.org/#/project"
Expand All @@ -37,18 +37,18 @@ class AssessmentKind(enum.Enum):
PAGE_SIZE = 100

# Put both bytes and str as keys for convenience.
CONTENT_TYPE_TO_EXT = {
CONTENT_TYPE_TO_EXT: dict[str | bytes, str] = {
"text/tab-separated-values": "tsv",
b"text/tab-separated-values": "tsv",
"application/vnd.ms-excel": "xls",
b"application/vnd.ms-excel": "xls",
}

EXT_TO_CONTENT_TYPE = dict(
EXT_TO_CONTENT_TYPE: dict[str, str] = dict(
(ext, ct) for ct, ext in CONTENT_TYPE_TO_EXT.items() if isinstance(ct, str)
)

WIKIDATA_PREFIXES = {
WIKIDATA_PREFIXES: dict[str, str] = {
"bd": "http://www.bigdata.com/rdf#",
"cc": "http://creativecommons.org/ns#",
"dct": "http://purl.org/dc/terms/",
Expand Down
8 changes: 5 additions & 3 deletions wp1/db.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,23 +3,25 @@
import time

import pymysql
import pymysql.connections
import pymysql.cursors
import pymysql.err
import socks

from wp1.credentials import CREDENTIALS, ENV
from wp1.environment import Environment

logger = logging.getLogger(__name__)

RETRY_TIME_SECONDS = 5


def connect(db_name, **overrides):
def connect(db_name: str, **overrides: object) -> pymysql.connections.Connection:
creds = CREDENTIALS[ENV].get(db_name)
if creds is None:
raise ValueError("db credentials for %r in ENV=%s are None" % (db_name, ENV))

kwargs = {
kwargs: dict[str, object] = {
"charset": None,
"use_unicode": False,
"cursorclass": pymysql.cursors.SSDictCursor,
Expand All @@ -30,7 +32,7 @@ def connect(db_name, **overrides):
tries = 4
while True:
try:
if db_name == "WIKIDB" and ENV == ENV.DEVELOPMENT:
if db_name == "WIKIDB" and ENV == Environment.DEVELOPMENT:
# In development, connect through a SOCKS5 proxy so that hosts on
# *.eqiad.wmflabs can be reached.
s = socks.socksocket()
Expand Down