"""Miscellaneous helper functions for mysql queries."""
#
# (C) Pywikibot team, 2016-2022
#
# Distributed under the terms of the MIT license.
#
from typing import Optional
import pkg_resources
import pywikibot
from pywikibot import config
from pywikibot.backports import removesuffix
try:
import pymysql
except ImportError:
raise ImportError('MySQL python module not found. Please install PyMySQL.')
[docs]def mysql_query(query: str, params=None,
dbname: Optional[str] = None,
verbose: Optional[bool] = None):
"""Yield rows from a MySQL query.
An example query that yields all ns0 pages might look like::
SELECT
page_namespace,
page_title,
FROM page
WHERE page_namespace = 0;
Supported MediaWiki projects use Unicode (UTF-8) character encoding.
Cursor charset is utf8.
:param query: MySQL query to execute
:param params: input parameters for the query, if needed
if list or tuple, %s shall be used as placeholder in the query string.
if a dict, %(key)s shall be used as placeholder in the query string.
:type params: tuple, list or dict of str
:param dbname: db name
:param verbose: if True, print query to be executed;
if None, config.verbose_output will be used.
:return: generator which yield tuples
"""
# These are specified in config.py or your user config file
if verbose is None:
verbose = config.verbose_output
if config.db_connect_file is None:
credentials = {'user': config.db_username,
'password': config.db_password}
else:
credentials = {'read_default_file': config.db_connect_file}
pymysql_version = pkg_resources.parse_version(
removesuffix(pymysql.__version__, '.None'))
args = {
'host': config.db_hostname_format.format(dbname),
'database': config.db_name_format.format(dbname),
'port': config.db_port,
'charset': 'utf8',
'defer_connect': query == 'test', # for tests
}
connection = pymysql.connect(**args, **credentials)
if pymysql_version < pkg_resources.parse_version('1.0.0'):
from contextlib import closing
connection = closing(connection)
with connection as conn, conn.cursor() as cursor:
if verbose:
_query = cursor.mogrify(query, params)
if not isinstance(_query, str):
_query = str(_query, encoding='utf-8')
_query = _query.strip()
_query = '\n'.join(f' {line}'
for line in _query.splitlines())
pywikibot.info('Executing query:\n' + _query)
if query == 'test': # for tests only
yield query
cursor.execute(query, params)
yield from cursor