787 lines
39 KiB
Python
787 lines
39 KiB
Python
from typing import Union, List, Optional, Tuple
|
|
from pathlib import Path
|
|
from enum import Enum
|
|
import json
|
|
import logging
|
|
import re
|
|
import os
|
|
import abc
|
|
import sqlite3
|
|
from .mysql2sqlite import mysql_to_sqlite
|
|
import subprocess
|
|
import traceback
|
|
import pexpect
|
|
|
|
|
|
SqlQuery = str
|
|
|
|
|
|
class Table:
|
|
'''a table is a list of rows, where each row is a tuple of column values
|
|
'''
|
|
column_names: Optional[List[str]]
|
|
rows: List[Tuple]
|
|
|
|
def __init__(self, column_names: Optional[List[str]], rows: List[Tuple]):
|
|
self.column_names = column_names
|
|
self.rows = rows
|
|
|
|
def get_value(self, row: int, col: Union[int, str]) -> Union[str, int, float]:
|
|
'''returns the value at the given row and column in this table
|
|
:param row: the index of the row (0-based)
|
|
:param col: the index or name of the column
|
|
'''
|
|
if isinstance(col, str):
|
|
assert self.column_names is not None, 'Cannot get value by column name because this table does not have column names'
|
|
try:
|
|
col_index = self.column_names.index(col)
|
|
except ValueError as exc:
|
|
raise ValueError(f'Column name "{col}" not found in table columns {self.column_names}') from exc
|
|
col = col_index
|
|
assert isinstance(col, int), f'Unexpected type for col: expected int but got {type(col)}'
|
|
assert 0 <= row < len(self.rows), f'Row index {row} is out of bounds for table with {len(self.rows)} rows'
|
|
assert 0 <= col < len(self.rows[row]), f'Column index {col} is out of bounds for table with {len(self.rows[row])} columns'
|
|
return self.rows[row][col]
|
|
|
|
def check_integrity(self):
|
|
'''checks that all rows in this table have the same number of columns and that the number of column names (if any) matches the number of columns in the rows
|
|
'''
|
|
num_cols = None
|
|
for row in self.rows:
|
|
if num_cols is None:
|
|
num_cols = len(row)
|
|
else:
|
|
assert len(row) == num_cols, f'All rows in the table should have the same number of columns but got a row with {len(row)} columns while previous rows have {num_cols} columns'
|
|
if self.column_names is not None:
|
|
assert len(self.column_names) == num_cols, f'Number of column names {len(self.column_names)} does not match number of columns in rows {num_cols}'
|
|
|
|
@property
|
|
def num_rows(self) -> int:
|
|
return len(self.rows)
|
|
|
|
@property
|
|
def num_cols(self) -> int:
|
|
nc = len(self.rows[0]) if len(self.rows) > 0 else 0
|
|
if self.column_names is not None:
|
|
assert len(self.column_names) == nc, f'Number of column names {len(self.column_names)} does not match number of columns in rows {nc}'
|
|
return nc
|
|
|
|
|
|
class SqlTableField():
|
|
'''description of a field of a sql table
|
|
'''
|
|
class Type(Enum):
|
|
FIELD_TYPE_STRING = 0
|
|
FIELD_TYPE_INT = 1
|
|
FIELD_TYPE_FLOAT = 2
|
|
FIELD_TYPE_TIME = 3
|
|
|
|
name: str # the name of the field, eg 'matrix_size'
|
|
field_type: Type # the type of the field, eg 'PARAM_TYPE_INT'
|
|
description: str # the description of the field, eg 'the size n of the n*n matrix '
|
|
is_autoinc_index: bool # indicates if theis field is used as an autoincrement index in the table
|
|
|
|
def __init__(self, name: str, field_type: Type, description: str, is_autoinc_index=False):
|
|
if is_autoinc_index:
|
|
assert field_type == SqlTableField.Type.FIELD_TYPE_INT, 'only an integer field can be used as a autoincrement table index'
|
|
self.name = name
|
|
self.field_type = field_type
|
|
self.description = description
|
|
self.is_autoinc_index = is_autoinc_index
|
|
|
|
|
|
ColumnId = str # eg 'matrix_size'
|
|
TableId = str # eg 'benchmark_results'
|
|
|
|
|
|
class ISqlConnection(object):
|
|
def __init__(self):
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def query(self, sql_query: SqlQuery) -> List[str]:
|
|
"""
|
|
:param str sql_query: the sql query to perform
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
def query_select(self, columns: List[ColumnId], table: TableId, where_clause: Optional[str] = None, join_clause: Optional[str] = None, distinct: bool = False) -> Table:
|
|
"""
|
|
performs a select query on the sql database and returns the results in the form of a list of tuples (one tuple per row, tuple values are the column values)
|
|
|
|
:param List[ColumnId] columns: the columns to select
|
|
:param TableId table: the name of the table to query
|
|
:param Optional[str] where_clause: the where clause for the query, eg "matrix_size > 100"
|
|
:param Optional[str] join_clause: the join clause for the query, eg "disables ON log.id = disables.disable_request_id"
|
|
:param bool distinct: whether to return distinct rows
|
|
:return: the results of the query
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
def query_insert(self, table_id: str, fields: List[str], values: List[tuple]) -> int:
|
|
"""
|
|
performs an insert query on the sql database and returns the id of the inserted row
|
|
|
|
:param str table_id: the name of the table to insert into
|
|
:param List[str] fields: the list of fields to insert values into
|
|
:param List[tuple] values: the list of values to insert (one tuple per row, tuple values are the column values)
|
|
:return: the id of the inserted row
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def table_exists(self, table_name: str) -> bool:
|
|
"""returns true if the given table exists in the database
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def create_table(self, table_name: str, fields: List[SqlTableField]):
|
|
"""creates the table in this sql database
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def delete_table(self, table_name: str):
|
|
"""deletes the given table in this sql database
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
# @abc.abstractmethod
|
|
# def add_table_row(self, table_name: str, values: List[SqlTableField]):
|
|
# """creates the table in this sql database
|
|
# """
|
|
# raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def get_field_directive(self, field_name: str, field_sql_type: str, field_description: str) -> str:
|
|
"""returns the sql directive for the declaration of the given table field (eg "`matrix_size` real NOT NULL COMMENT 'the size of the matrix'")
|
|
"""
|
|
raise NotImplementedError()
|
|
|
|
@abc.abstractmethod
|
|
def dump(self, sql_file_path: Path):
|
|
"""dumps this database into the given sql file"""
|
|
raise NotImplementedError()
|
|
|
|
|
|
class ISqlDatabaseBackend(object):
|
|
def __init__(self):
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def connect(self) -> ISqlConnection:
|
|
"""returns a connection to this sql database backend that can be used to perform sql queries on the database"""
|
|
raise NotImplementedError()
|
|
|
|
|
|
def values_to_sql_string(values: List[tuple]) -> str:
|
|
'''converts a list of tuples of values into a string that can be used in a sql query, with proper escaping of string values
|
|
|
|
eg the list of tuples
|
|
[('alambix42', 'disabled to move the alambix42 to another rack'), ('alambix42', 'because I want to test quman')]
|
|
will be converted into the following string that can be used in a sql query :
|
|
"('alambix42', 'disabled to move the alambix42 to another rack'), ('alambix42', 'because I want to test quman')"
|
|
'''
|
|
sql_values = []
|
|
for value_tuple in values:
|
|
escaped_value_tuple = tuple(["'%s'" % str(v).replace("'", r"\'") for v in value_tuple])
|
|
sql_values.append(f'({", ".join(escaped_value_tuple)})')
|
|
return ', '.join(sql_values)
|
|
|
|
|
|
# class RemoteMysqlDb(ISqlDatabaseBackend):
|
|
# def __init__(self, db_server_fqdn, db_user, db_name):
|
|
# """
|
|
# :param str db_server_fqdn: the fully qualified domain name of the server hosting the database, eg simpatix10.univ-rennes1.fr
|
|
# :param str db_user: the user for accessing the inventory database, eg simpadb_reader
|
|
# :param str db_name: the name of the inventory database, eg simpadb
|
|
# """
|
|
# self.db_server_fqdn = db_server_fqdn
|
|
# self.db_user = db_user
|
|
# self.db_name = db_name
|
|
# self._connect()
|
|
|
|
# def _connect(self):
|
|
# self._conn = MySQLdb.connect(self.db_server_fqdn, self.db_user, '', self.db_name)
|
|
# assert self._conn
|
|
|
|
# def query(self, sql_query) -> List[str]:
|
|
# """
|
|
# :param str sql_query: the sql query to perform
|
|
# """
|
|
# cursor = self._conn.cursor()
|
|
# cursor.execute(sql_query)
|
|
# rows = cursor.fetchall()
|
|
# logging.debug("RemoteMysqlDb.query:: results of query using cursor '%s': %s", sql_query, rows)
|
|
# return rows
|
|
|
|
# def query_select(self, columns: List[str], table: str, where_clause: Optional[str] = None, join_clause: Optional[str] = None, distinct: bool = False) -> Table:
|
|
# sql_query = f"SELECT {('DISTINCT ' if distinct else '')}{', '.join(columns)} FROM {table}"
|
|
# if join_clause:
|
|
# sql_query += f" JOIN {join_clause}"
|
|
# if where_clause:
|
|
# sql_query += f" WHERE {where_clause}"
|
|
# return self.query(sql_query)
|
|
|
|
# def query_insert(self, table_id: str, fields: List[str], values: List[tuple]) -> int:
|
|
# sql_query = f"INSERT INTO {table_id} ({', '.join(fields)}) VALUES {values_to_sql_string(values)}"
|
|
# sql_query = sql_query + ";SELECT last_insert_id();"
|
|
# logging.debug("RemoteMysqlDb.query_insert:: sql_query = '%s'", sql_query)
|
|
# rows = self.query(sql_query)
|
|
# last_insert_id = int(rows[0][0])
|
|
# assert last_insert_id > 0, f'Unexpected last insert id : {last_insert_id}'
|
|
# return last_insert_id
|
|
|
|
# def table_exists(self, table_name: str) -> bool:
|
|
# rows = self.query(f"SHOW TABLES LIKE '{table_name}';")
|
|
# assert len(rows) <= 1, f'Unexpected case: more than one ({len(rows)}) tables match the table name {table_name}.'
|
|
# return len(rows) == 1
|
|
|
|
# def create_table(self, table_name: str, fields: List[SqlTableField]):
|
|
# raise NotImplementedError()
|
|
|
|
# def get_field_directive(self, field_name: str, field_sql_type: str, field_description: str) -> str:
|
|
# return f'`{field_name}` {field_sql_type} COMMENT \'{field_description}\''
|
|
|
|
# def dump(self, sql_file_path: Path):
|
|
# raise NotImplementedError()
|
|
|
|
|
|
def json_to_table(json_str: str) -> Table:
|
|
assert isinstance(json_str, str), f'Expected a string as input for json_to_table but got {type(json_str)}' # eg '[{"toto": "2026-05-21 15:11:58", "iddd": "graffy"},{"toto": "2026-05-22 15:55:39", "iddd": "graffy"}]'
|
|
# logging.debug("json_to_table:: json_str = '%s'", json_str)
|
|
if json_str == 'NULL':
|
|
return Table(column_names=None, rows=[])
|
|
try:
|
|
json_data = json.loads(json_str)
|
|
except json.JSONDecodeError:
|
|
logging.error("json_to_table:: invalid json string: '%s'", json_str)
|
|
raise
|
|
assert isinstance(json_data, list), f'Expected a list of rows in the json string but got {type(json_data)}'
|
|
table = []
|
|
for row in json_data:
|
|
row_values = []
|
|
for value in row.values():
|
|
assert isinstance(value, (str, int, float)), f'Expected a string, int or float value for each column value in the json string but got {type(value)}'
|
|
row_values.append(value)
|
|
table.append(tuple(row_values))
|
|
return Table(column_names=None, rows=table)
|
|
|
|
|
|
def print_line_chars(line: str):
|
|
for char in line:
|
|
# char identifier, eg 'ESC' for 0x1b
|
|
char_name = {
|
|
'\x1b': 'ESC',
|
|
'\r': 'CR',
|
|
'\n': 'LF'
|
|
}.get(char, repr(char))
|
|
logging.debug('char = \'%s\' (hex=%s)', char_name, char.encode("utf-8").hex())
|
|
|
|
|
|
def strip_vt100(text):
|
|
# ANSI: Standardized, starts with \x1b[.
|
|
# VT100: Includes private sequences like \x1b(B, \x1b(A, \x1b>, etc., which are not ANSI but widely supported in terminal emulators.
|
|
# Matches all VT100 escape sequences:
|
|
# - CSI: \x1b[... (e.g., \x1b[0;1m)
|
|
# - Fe: \x1b(..., \x1b), \x1b> (e.g., \x1b(B, \x1b>A)
|
|
# - Single-character: \x1b[@-Z\\_-]
|
|
vt100_escape = re.compile(r'\x1b(?:[@-Z\\_-]|\[[0-?]*[ -/]*[@-~]|\([A-Za-z]|\)[A-Za-z])')
|
|
return vt100_escape.sub('', text)
|
|
|
|
|
|
def parse_mysql_stdout(stdout: str) -> Table:
|
|
# example of mysql query output for the query "SHOW TABLES LIKE 'log';":
|
|
# > SHOW TABLES LIKE 'log';
|
|
# > +-----------------------+
|
|
# > | Tables_in_quman (log) |
|
|
# > +-----------------------+
|
|
# > | log |
|
|
# > +-----------------------+
|
|
# > 1 row in set (0,001 sec)
|
|
# >
|
|
# >
|
|
# logging.debug("parse_mysql_stdout:: stdout = '%s'", stdout)
|
|
table = []
|
|
line_index = 0
|
|
num_horizontal_separators = 0
|
|
num_rows = None
|
|
num_empty_lines = 0
|
|
column_names = None
|
|
for line in stdout.split('\n'):
|
|
# print_line_chars(line)
|
|
line = strip_vt100(line)
|
|
line = line.strip('\r') # remove potential carriage return characters that can be present in the mysql output
|
|
# remove all ansi escape codes
|
|
# logging.debug("parse_mysql_stdout:: line = '%s' (%d characters, hex=%s)", line, len(line), line.encode('utf-8').hex())
|
|
if re.match(r'^\s*\+\-+\+\s*$', line):
|
|
num_horizontal_separators += 1
|
|
elif re.match(r'^\|.*\|$', line):
|
|
# eg '| Tables_in_quman (log) |'
|
|
is_header = num_horizontal_separators == 1
|
|
if is_header:
|
|
column_names = [col.strip() for col in line.split('|')[1:-1]]
|
|
else:
|
|
assert num_horizontal_separators == 2, f'Unexpected case: a line with values should be between two horizontal separators but got num_horizontal_separators={num_horizontal_separators} for line="{line}"'
|
|
values = [line.strip() for line in line.split('|')[1:-1]]
|
|
table.append(tuple(values))
|
|
elif line.endswith(';'):
|
|
# this is the echo of the query that we sent to mysql, we can ignore it but we check that it is indeed the first line of the output
|
|
# eg: 'SHOW TABLES LIKE 'log';'
|
|
assert line_index == 0, f'Unexpected case: the first line of the mysql output should be the query echo but got "{line}"'
|
|
elif re.match(r'^Empty set \([^\)]*\)$', line):
|
|
# eg: 'Empty set (0,000 sec)'
|
|
num_rows = 0
|
|
elif re.match(r'^Query OK', line):
|
|
# eg Query OK, 1 row affected (0,005 sec)
|
|
pass
|
|
else:
|
|
m = re.match(r'^\s*(?P<num_rows>\d+)\s+row[s]*\s+in\s+set\s+\(.*\)\s*$', line)
|
|
if m:
|
|
# eg '1 row in set (0,001 sec)'
|
|
num_rows = int(m.group('num_rows'))
|
|
elif line.strip() == '':
|
|
num_empty_lines += 1
|
|
else:
|
|
assert False, f'Unexpected line in mysql output: "{line}"'
|
|
line_index += 1
|
|
if num_rows is None:
|
|
num_rows = len(table)
|
|
assert num_rows == len(table), f'Unexpected case: the number of rows in the mysql output should be {num_rows} but got {len(table)}.'
|
|
if num_rows > 0:
|
|
assert num_horizontal_separators == 3, f'Unexpected case: there should be exactly 3 horizontal separators in the mysql output but got {num_horizontal_separators}.'
|
|
return Table(column_names=column_names, rows=table)
|
|
|
|
|
|
class SshAccessedMysqlConnection(ISqlConnection):
|
|
|
|
ssh_mysql_db: 'SshAccessedMysqlDb'
|
|
mysql_spawn: Optional[pexpect.spawn] = None
|
|
mysql_prompt_re: str
|
|
|
|
def __init__(self, ssh_mysql_db: 'SshAccessedMysqlDb'):
|
|
self.ssh_mysql_db = ssh_mysql_db
|
|
self.mysql_spawn = None
|
|
self.mysql_prompt_re = r'MariaDB \[[^]]+\]> '
|
|
|
|
def __enter__(self) -> 'SshAccessedMysqlConnection':
|
|
# Set up the connection (e.g., open SSH tunnel, connect to MySQL)
|
|
|
|
command = f'ssh -t "{self.ssh_mysql_db.ssh_user}@{self.ssh_mysql_db.db_server_fqdn}" "mysql --defaults-group-suffix={self.ssh_mysql_db.db_user}"'
|
|
# command = [
|
|
# 'ssh',
|
|
# '-t',
|
|
# f"{self.ssh_mysql_db.ssh_user}@{self.ssh_mysql_db.db_server_fqdn}",
|
|
# f'mysql --defaults-group-suffix={self.ssh_mysql_db.db_user}'
|
|
# ]
|
|
|
|
logging.debug("SshAccessedMysqlConnection.__enter__: running ssh command '%s'", command)
|
|
try:
|
|
self.mysql_spawn = pexpect.spawn(command, encoding='utf-8', timeout=10)
|
|
assert isinstance(self.mysql_spawn, pexpect.spawn)
|
|
# graffy@alambix50:~$ /usr/bin/ssh -t qumandbw@alambix-master.ipr.univ-rennes.fr 'mysql --defaults-group-suffix=qumandbw'
|
|
# Welcome to the MariaDB monitor. Commands end with ; or \g.
|
|
# Your MariaDB connection id is 845
|
|
# Server version: 10.11.14-MariaDB-0+deb12u2 Debian 12
|
|
|
|
# Copyright (c) 2000, 2018, Oracle, MariaDB Corporation Ab and others.
|
|
|
|
# Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.
|
|
|
|
# MariaDB [(none)]>
|
|
self.mysql_spawn.expect(self.mysql_prompt_re) # Wait for MySQL prompt
|
|
|
|
show_databases = False
|
|
if show_databases:
|
|
self.mysql_spawn.sendline("show databases;")
|
|
self.mysql_spawn.expect(self.mysql_prompt_re)
|
|
logging.debug("databases:")
|
|
logging.debug(self.mysql_spawn.before) # Print output
|
|
|
|
self.mysql_spawn.sendline(f"use {self.ssh_mysql_db.db_name};")
|
|
self.mysql_spawn.expect(self.mysql_prompt_re)
|
|
|
|
except ValueError as e:
|
|
logging.error("failed to execute the command '%s' (error: %s)", command, str(e))
|
|
traceback.print_exc()
|
|
raise RuntimeError(f"Error while connecting to mysql: {str(e)}") from e
|
|
return self # Return the connection object for use in the 'with'
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
# Clean up the connection
|
|
logging.debug("SshAccessedMysqlConnection.__exit__: Closing connection...")
|
|
self.mysql_spawn.sendline("quit")
|
|
self.mysql_spawn.expect(pexpect.EOF)
|
|
self.mysql_spawn = None
|
|
if exc_type:
|
|
logging.error("An error occurred: %s", exc_val)
|
|
return False # Propagate exceptions
|
|
|
|
def query(self, sql_query: SqlQuery) -> List[str]:
|
|
"""
|
|
:param str sql_query: the sql query to perform
|
|
"""
|
|
try:
|
|
self.mysql_spawn.sendline(sql_query)
|
|
self.mysql_spawn.expect(self.mysql_prompt_re)
|
|
|
|
stdout = self.mysql_spawn.before
|
|
except pexpect.exceptions.ExceptionPexpect as e:
|
|
logging.error("SshAccessedMysqlDb.query:: error while executing query '%s': %s", sql_query, str(e))
|
|
raise RuntimeError(f"Error while executing query '{sql_query}': {e}") from e
|
|
logging.debug("SshAccessedMysqlDb.query:: results of query '%s': %s", sql_query, stdout)
|
|
rows = stdout.split('\n') if stdout != '' else []
|
|
# detect errors in the output such as:
|
|
# ERROR 1142 (42000): DELETE command denied to user 'qumandbw'@'localhost' for table `quman`.`disables`
|
|
for row in rows:
|
|
m = re.match(r'^ERROR ', row)
|
|
if m is not None:
|
|
raise RuntimeError(f'the sql query "{sql_query}" failed with the error : "{row}"')
|
|
return rows
|
|
|
|
def query_select(self, columns: List[str], table: str, where_clause: Optional[str] = None, join_clause: Optional[str] = None, distinct: bool = False) -> Table:
|
|
# MariaDB [quman]> SELECT JSON_ARRAYAGG(JSON_OBJECT('toto', timestamp, 'iddd', user_id)) FROM log;
|
|
# +-------------------------------------------------------------------------------------------------------+
|
|
# | JSON_ARRAYAGG(JSON_OBJECT('toto', timestamp, 'iddd', user_id)) |
|
|
# +-------------------------------------------------------------------------------------------------------+
|
|
# | [{"toto": "2026-05-21 15:11:58", "iddd": "graffy"},{"toto": "2026-05-22 15:55:39", "iddd": "graffy"}] |
|
|
# +-------------------------------------------------------------------------------------------------------+
|
|
# 1 row in set (0,006 sec)
|
|
# columns_list_str = ', '.join([f'"{col}"' for col in columns])
|
|
columns_list_str = ', '.join([f"'{col}', {col}" for col in columns])
|
|
columns_statement = f'JSON_ARRAYAGG(JSON_OBJECT({columns_list_str}))'
|
|
sql_query = f"SELECT {('DISTINCT ' if distinct else '')}{columns_statement} FROM {table}"
|
|
if join_clause:
|
|
sql_query += f" JOIN {join_clause}"
|
|
if where_clause:
|
|
sql_query += f" WHERE {where_clause}"
|
|
sql_query += ';'
|
|
lines = self.query(sql_query)
|
|
table = parse_mysql_stdout('\n'.join(lines))
|
|
assert table.num_cols == 1, f'Unexpected case: the query should return only one column with the json data but got {table.num_cols} columns.'
|
|
assert table.num_rows == 1, f'Unexpected case: the query should return only one row with the json data but got {table.num_rows} rows.'
|
|
json_str = table.get_value(row=0, col=0)
|
|
# json_str is expected to contain a string like '[{"toto": "2026-05-21 15:11:58", "iddd": "graffy"},{"toto": "2026-05-22 15:55:39", "iddd": "graffy"}]'
|
|
assert isinstance(json_str, str), f'Expected a string as the value of the only cell in the table output but got {type(json_str)}'
|
|
# logging.debug("SshAccessedMysqlDb.query_select:: value (in the form of a stringified json tree) of the cell for the query '%s': '%s'", sql_query, json_str)
|
|
# match = re.match(r'^\s*(?P<json_data>{[^}]*})\s*$', data_line)
|
|
# assert match, 'Unexpected output format for query "%s" : %s' % (sql_query, stdout)
|
|
# json_data = json.loads(match.group('json_data'))
|
|
t = json_to_table(json_str)
|
|
assert isinstance(t, Table), f'Unexpected case: json_to_table should return a Table object but got {type(t)}'
|
|
return t
|
|
|
|
def query_insert(self, table_id: str, fields: List[str], values: List[tuple]) -> int:
|
|
logging.debug("SshAccessedMysqlDb.query_insert:: values = %s", values)
|
|
# INSERT INTO log (timestamp, user_id, host_fqdn, queue_machines, action, disable_id, reason) VALUES ('2026-05-26T15:31:53.564287', 'graffy', 'alambix50.ipr.univ-rennes.fr', 'gpuonly.q@alambix104.ipr.univ-rennes1.fr', 'disable', 'quman-sync', 'synchronized with grid engine');SELECT JSON_OBJECT('toto', last_insert_id());
|
|
# Query OK, 1 row affected (0,046 sec)
|
|
|
|
# +---------------------------------------+
|
|
# | JSON_OBJECT('toto', last_insert_id()) |
|
|
# +---------------------------------------+
|
|
# | {"toto": 9} |
|
|
# +---------------------------------------+
|
|
# 1 row in set (0,002 sec)
|
|
sql_query = f"INSERT INTO {table_id} ({', '.join(fields)}) VALUES {values_to_sql_string(values)}"
|
|
sql_query = sql_query + ";SELECT JSON_OBJECT('toto', last_insert_id());"
|
|
logging.debug("SshAccessedMysqlDb.query_insert:: sql_query = '%s'", sql_query)
|
|
stdout_lines = self.query(sql_query)
|
|
table: Table = parse_mysql_stdout('\n'.join(stdout_lines))
|
|
assert table.num_cols == 1, f'Unexpected case: the query should return only one column with the json data but got {table.num_cols} columns.'
|
|
assert table.num_rows == 1, f'Unexpected case: the query should return only one row with the json data but got {table.num_rows} rows.'
|
|
data_line = table.get_value(row=0, col=0)
|
|
assert isinstance(data_line, str), f'Expected a string as data line in the query output but got {type(data_line)}'
|
|
logging.debug("SshAccessedMysqlDb.query_insert:: data line of query '%s': '%s'", sql_query, data_line)
|
|
match = re.match(r'^\s*(?P<json_data>{\s*"toto"\s*:\s*(\d+)\s*})\s*$', data_line)
|
|
assert match, 'Unexpected output format for query "%s" : %s' % (sql_query, data_line)
|
|
json_data = json.loads(match.group('json_data'))
|
|
last_insert_id = int(json_data['toto'])
|
|
assert last_insert_id > 0, f'Unexpected last insert id : {last_insert_id}'
|
|
return last_insert_id
|
|
|
|
def table_exists(self, table_name: str) -> bool:
|
|
rows = self.query(f"SHOW TABLES LIKE '{table_name}';")
|
|
table: Table = parse_mysql_stdout('\n'.join(rows))
|
|
logging.debug('len(rows) = %d', len(rows))
|
|
logging.debug('rows = %s', str(rows))
|
|
assert table.num_rows <= 1, f'Unexpected case: more than one ({len(table)}) tables match the table name {table_name}.'
|
|
if table.num_rows > 0:
|
|
assert table.column_names[0].startswith('Tables_in_'), f'Unexpected case: the column name in the output of "SHOW TABLES LIKE ..." should start with "Tables_in_" but got "{table.column_names[0]}"'
|
|
assert table.get_value(row=0, col=0) == table_name, f'Unexpected case: the table name in the output of "SHOW TABLES LIKE ..." should be "{table_name}" but got "{table.get_value(row=0, col=0)}"'
|
|
return table.num_rows == 1
|
|
|
|
def create_table(self, table_name: str, fields: List[SqlTableField]):
|
|
# https://www.sqlite.org/autoinc.html
|
|
# > The AUTOINCREMENT keyword imposes extra CPU, memory, disk space, and disk I/O overhead and should be avoided if not strictly needed. It is usually not needed.
|
|
fields_sql_descriptions = []
|
|
for field in fields:
|
|
sql_field_type = {
|
|
SqlTableField.Type.FIELD_TYPE_FLOAT: 'real NOT NULL',
|
|
SqlTableField.Type.FIELD_TYPE_INT: 'int(11) NOT NULL',
|
|
SqlTableField.Type.FIELD_TYPE_STRING: 'varchar(256) NOT NULL',
|
|
SqlTableField.Type.FIELD_TYPE_TIME: 'datetime NOT NULL',
|
|
}[field.field_type]
|
|
if field.is_autoinc_index:
|
|
assert field.field_type == SqlTableField.Type.FIELD_TYPE_INT
|
|
sql_field_type = 'INTEGER PRIMARY KEY AUTO_INCREMENT'
|
|
fields_sql_description = self.get_field_directive(field.name, sql_field_type, field.description)
|
|
fields_sql_descriptions.append(fields_sql_description)
|
|
|
|
sql_create_table_command = f'CREATE TABLE `{table_name}` ({",".join(fields_sql_descriptions)});'
|
|
logging.debug('sql_create_table_command = %s', sql_create_table_command)
|
|
self.query(sql_create_table_command)
|
|
|
|
def delete_table(self, table_name: str):
|
|
sql_create_table_command = f'DROP TABLE `{table_name}`;'
|
|
self.query(sql_create_table_command)
|
|
|
|
def get_field_directive(self, field_name: str, field_sql_type: str, field_description: str) -> str:
|
|
return f'`{field_name}` {field_sql_type} COMMENT \'{field_description}\''
|
|
|
|
def dump(self, sql_file_path: Path):
|
|
# mysqldump -u root --quote-names --opt --single-transaction --quick $db >
|
|
db = self.ssh_mysql_db
|
|
command = f'ssh "{db.ssh_user}@{db.db_server_fqdn}" "mysqldump --defaults-group-suffix={db.db_user}" {db.db_name} > {sql_file_path}'
|
|
_ = subprocess.run(command, shell=True, check=False, capture_output=True)
|
|
|
|
|
|
class SshAccessedMysqlDb(ISqlDatabaseBackend):
|
|
"""a mysql database server accessed using ssh instead of a remote mysql client
|
|
|
|
Instead of accessing the remote sql database from mysql client, this method ssh connects (we expect an unattended connection setup using ssh keys) to the server hosting the database, and once logged in performs a request to the database locally using mysql client running on the server hosting the database.
|
|
|
|
This method has a the following benefits over the simple use of mysql:
|
|
- it's more secure since it benefits from ssh encryption
|
|
- it doesn't require extra open tcp ports for remote mysql database access
|
|
|
|
"""
|
|
db_server_fqdn: str # the fully qualified domain name of the server hosting the database, eg iprbenchdb.ipr.univ-rennes1.fr
|
|
db_user: str # the user for accessing the inventory database, eg iprbenchw
|
|
db_name: str # the name of the database, eg iprbench
|
|
ssh_user: str # the user on db_server_fqdn that has access to the database db_name
|
|
|
|
def __init__(self, db_server_fqdn: str, db_user: str, db_name: str, ssh_user: str):
|
|
"""
|
|
:param str db_server_fqdn: the fully qualified domain name of the server hosting the database, eg iprbenchdb.ipr.univ-rennes1.fr
|
|
:param str db_user: the user for accessing the inventory database, eg iprbenchw
|
|
:param str db_name: the name of the database, eg iprbench
|
|
:param str ssh_user: the user on db_server_fqdn that has access to the database db_name
|
|
"""
|
|
self.db_server_fqdn = db_server_fqdn
|
|
self.db_user = db_user
|
|
self.db_name = db_name
|
|
self.ssh_user = ssh_user
|
|
|
|
def connect(self) -> SshAccessedMysqlConnection:
|
|
|
|
return SshAccessedMysqlConnection(self)
|
|
|
|
|
|
class SqliteConnection(ISqlConnection):
|
|
sqlite_db: 'SqliteDb'
|
|
_con: sqlite3.Connection
|
|
_cur: sqlite3.Cursor
|
|
|
|
def __init__(self, sql_db: 'SqliteDb'):
|
|
self.sqlite_db = sql_db
|
|
self._con = None
|
|
self._cur = None
|
|
|
|
def __enter__(self) -> 'SqliteConnection':
|
|
# Set up the connection (e.g., open SSH tunnel, connect to MySQL)
|
|
logging.debug("Opening connection...")
|
|
|
|
check_same_thread = False
|
|
# this is to prevent the following error when run from apache/django : SQLite objects created in a thread can only be used in that same thread.The object was created in thread id 139672342353664 and this is thread id 139672333960960
|
|
# accordig to https://stackoverflow.com/questions/48218065/programmingerror-sqlite-objects-created-in-a-thread-can-only-be-used-in-that-sa this is ok, as long as there are no concurrent writes
|
|
# If set False, the returned connection may be shared across multiple threads. When using multiple threads with the same connection writing operations should be serialized by the user to avoid data corruption
|
|
# I hope it's safe here but I'm not 100% sure though. Anyway, if the database gets corrupt, it not a big deal since this memory resident database gets reconstructed from the sql file...
|
|
sqlite_db_path = self.sqlite_db.sqlite_db_path
|
|
if sqlite_db_path == ':memory:' or not sqlite_db_path.exists():
|
|
logging.debug('creating sqlite database in %s', sqlite_db_path)
|
|
self._con = sqlite3.connect(sqlite_db_path, check_same_thread=check_same_thread)
|
|
else:
|
|
logging.debug('reusing existing sqlite database in %s', sqlite_db_path)
|
|
self._con = sqlite3.connect(sqlite_db_path, check_same_thread=check_same_thread)
|
|
self._cur = self._con.cursor()
|
|
logging.debug('self._con = %s', self._con)
|
|
logging.debug('self._cur = %s', self._cur)
|
|
|
|
_ = self.query('PRAGMA encoding="UTF-8";')
|
|
|
|
return self # Return the connection object for use in the 'with'
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
# Clean up the connection
|
|
logging.debug("Closing connection...")
|
|
if exc_type:
|
|
logging.error("An error occurred: %s", exc_val)
|
|
return False # Propagate exceptions
|
|
|
|
def query(self, sql_query) -> List[str]:
|
|
"""
|
|
:param str sql_query: the sql query to perform
|
|
"""
|
|
self._cur.execute(sql_query)
|
|
rows = self._cur.fetchall()
|
|
self._con.commit()
|
|
return rows
|
|
|
|
def query_select(self, columns: List[str], table: str, where_clause: Optional[str] = None, join_clause: Optional[str] = None, distinct: bool = False) -> Table:
|
|
sql_query = f"SELECT {('DISTINCT ' if distinct else '')}{', '.join(columns)} FROM {table}"
|
|
if join_clause:
|
|
sql_query += f" JOIN {join_clause}"
|
|
if where_clause:
|
|
sql_query += f" WHERE {where_clause}"
|
|
rows = self.query(sql_query)
|
|
return Table(column_names=columns, rows=rows)
|
|
|
|
def query_insert(self, table_id: str, fields: List[str], values: List[tuple]) -> int:
|
|
"""
|
|
performs an insert query on the sql database and returns the id of the inserted row
|
|
|
|
:param str table_id: the name of the table to insert into
|
|
:param List[str] fields: the list of fields to insert values into
|
|
:param List[tuple] values: the list of values to insert (one tuple per row, tuple values are the column values)
|
|
:return: the id of the inserted row
|
|
"""
|
|
sql_query = f"INSERT INTO {table_id} ({', '.join(fields)}) VALUES {values_to_sql_string(values)}"
|
|
logging.debug("SqliteDb.query_insert:: sql_query = '%s'", sql_query)
|
|
logging.debug("SqliteDb.query_insert:: values = %s", values)
|
|
self._cur.execute(sql_query)
|
|
self._cur.execute('SELECT last_insert_rowid();')
|
|
rows = self._cur.fetchall()
|
|
self._con.commit()
|
|
assert len(rows) == 1, f'Unexpected number of rows ({len(rows)}).'
|
|
last_insert_id = int(rows[0][0])
|
|
assert last_insert_id > 0, f'Unexpected last insert id : {last_insert_id}'
|
|
return last_insert_id
|
|
|
|
def table_exists(self, table_name: str) -> bool:
|
|
rows = self.query(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';")
|
|
assert len(rows) <= 1, f'Unexpected case: more than one ({len(rows)}) tables match the table name {table_name}.'
|
|
return len(rows) == 1
|
|
|
|
def create_table(self, table_name: str, fields: List[SqlTableField]):
|
|
# https://www.sqlite.org/autoinc.html
|
|
# > The AUTOINCREMENT keyword imposes extra CPU, memory, disk space, and disk I/O overhead and should be avoided if not strictly needed. It is usually not needed.
|
|
fields_sql_descriptions = []
|
|
for field in fields:
|
|
sql_field_type = {
|
|
SqlTableField.Type.FIELD_TYPE_FLOAT: 'real NOT NULL',
|
|
SqlTableField.Type.FIELD_TYPE_INT: 'int(11) NOT NULL',
|
|
SqlTableField.Type.FIELD_TYPE_STRING: 'varchar(256) NOT NULL',
|
|
SqlTableField.Type.FIELD_TYPE_TIME: 'datetime NOT NULL',
|
|
}[field.field_type]
|
|
if field.is_autoinc_index:
|
|
assert field.field_type == SqlTableField.Type.FIELD_TYPE_INT
|
|
sql_field_type = 'INTEGER PRIMARY KEY'
|
|
fields_sql_description = self.get_field_directive(field.name, sql_field_type, field.description)
|
|
fields_sql_descriptions.append(fields_sql_description)
|
|
|
|
sql_create_table_command = f'CREATE TABLE `{table_name}` ({",".join(fields_sql_descriptions)});'
|
|
logging.debug('sql_create_table_command = %s', sql_create_table_command)
|
|
self.query(sql_create_table_command)
|
|
|
|
def delete_table(self, table_name: str):
|
|
sql_create_table_command = f'DROP TABLE `{table_name}`;'
|
|
self.query(sql_create_table_command)
|
|
|
|
def get_field_directive(self, field_name: str, field_sql_type: str, field_description: str) -> str:
|
|
# sqlite doesn't understand the COMMENT keyword, so we use sql comments ( "--" ), as explained in [https://stackoverflow.com/questions/7426205/sqlite-adding-comment-on-descriptions-to-tables-and-columns]
|
|
return f'`{field_name}` {field_sql_type} -- {field_description}\n'
|
|
|
|
def dump(self, sql_file_path: Path):
|
|
with open(sql_file_path, 'wt', encoding='utf8') as f:
|
|
for line in self._con.iterdump():
|
|
f.write(line)
|
|
|
|
|
|
class SqliteDb(ISqlDatabaseBackend):
|
|
sqlite_db_path: Union[Path, str] # the path of the sqlite database. ':memory:' # sqlite-specific special name for a file stored in memory. We could use something like '/tmp/simpadb.sqlite' here but this would make parsing really slow (1 minute instead of 1s), unless either :
|
|
|
|
def __init__(self, sqlite_db_path: Path):
|
|
self.sqlite_db_path = sqlite_db_path
|
|
|
|
def connect(self) -> SqliteConnection:
|
|
|
|
return SqliteConnection(self)
|
|
|
|
|
|
class SqlFile(SqliteDb):
|
|
_sql_file_path: Path
|
|
|
|
def __init__(self, sql_file_path: Path, truncate_hex_strings=False):
|
|
"""
|
|
:param str sql_file_path: the path of the sql file containing the inventory database
|
|
"""
|
|
self._sql_file_path = sql_file_path
|
|
# _con: sqlite3.Connection = None
|
|
# _cur: sqlite3.Cursor = None
|
|
|
|
sqlite_db_path = ':memory:' # sqlite-specific special name for a file stored in memory. We could use something like '/tmp/simpadb.sqlite' here but this would make parsing really slow (1 minute instead of 1s), unless either :
|
|
# - proper fix : group of INSERT statements are surrounded by BEGIN and COMMIT (see http://stackoverflow.com/questions/4719836/python-and-sqlite3-adding-thousands-of-rows)
|
|
# - the file is stored on a solid state disk
|
|
try:
|
|
os.remove(sqlite_db_path)
|
|
except BaseException: # pylint: disable=broad-except
|
|
pass
|
|
super().__init__(sqlite_db_path)
|
|
with open(str(self._sql_file_path), 'r', encoding='utf8') as f: # str conversion has been added to support older versions of python in which open don't accept arguments of type Path
|
|
sql = f.read() # watch out for built-in `str`
|
|
# print(sql)
|
|
|
|
sqlite_sql = mysql_to_sqlite(sql, truncate_hex_strings)
|
|
# print(mysql_to_sqlite(sql))
|
|
# with open('/tmp/toto.sqlite.sql', 'w') as f:
|
|
# f.write(sqlite_sql)
|
|
# with open('/tmp/toto.sqlite.sql', 'r') as f:
|
|
# sqlite_sql = f.read()
|
|
with self.connect() as conn:
|
|
conn._cur.executescript(sqlite_sql)
|
|
|
|
|
|
class TableAttrNotFound(Exception):
|
|
def __init__(self, table, key_name, key_value, attr_name):
|
|
message = "failed to find in table %s a value for %s where %s is %s" % (table, attr_name, key_name, key_value)
|
|
super(TableAttrNotFound, self).__init__(message)
|
|
self.table = table
|
|
self.key_name = key_name
|
|
self.key_value = key_value
|
|
self.attr_name = attr_name
|
|
|
|
|
|
class SqlDatabaseReader(object):
|
|
|
|
def __init__(self, inv_provider: ISqlDatabaseBackend):
|
|
"""
|
|
:param ISqlDatabaseBackend inv_provider: the input that provides the inventory data
|
|
"""
|
|
self._inv_provider = inv_provider
|
|
|
|
def query(self, sql_query: SqlQuery):
|
|
"""
|
|
performs a query on the sql database
|
|
|
|
:param SqlQuery sql_query: the sql query to perform
|
|
"""
|
|
return self._inv_provider.query(sql_query)
|
|
|
|
def get_table_attr(self, table, key_name, key_value, attr_name):
|
|
"""
|
|
reads the value of the fiven attribute of the given item in the given table
|
|
|
|
:param str table: the name of the table to read
|
|
:param str key_name: the name of the column that stores the id of the item to read
|
|
:param str key_value: the id of the item to read
|
|
:param str attr_name: the name of the attribute to read from the item
|
|
"""
|
|
attr_value = None
|
|
rows = self.query("SELECT " + attr_name + " FROM " + table + " WHERE " + key_name + "='" + key_value + "'")
|
|
if len(rows) > 0:
|
|
attr_value = rows[0][0]
|
|
else:
|
|
raise TableAttrNotFound(table, key_name, key_value, attr_name)
|
|
return attr_value
|