"""MySQL Client for ows-assets testing..""" from typing import Any from pymysql import connect from pymysql.cursors import DictCursor class MySQLClient: """MySQL Client for ows-assets testing.""" def __init__( self, host: str | None, db: str | None, user: str | None, password: str | None ) -> None: """Initialize MySQLClient class.""" self.db_host = host self.db_user = user self.db_password = password self.db_database = db def execute_query(self, query_string: str, parameters: Any) -> dict[str, Any]: """Execute a query, return first row and rows affected.""" connection = connect( host=self.db_host, user=self.db_user, password=self.db_password or "", database=self.db_database, ) try: cursor = connection.cursor(DictCursor) rows_affected = cursor.execute(query_string, parameters) finally: fetch_result = list(cursor.fetchall()) cursor.close() connection.commit() connection.close() return {"rows": fetch_result, "rows_affected": rows_affected}