Due to a scheduled upgrade to version 14.10, GitLab will be unavailabe on Monday 30.05., from 19:00 until 20:00.

Unverified Commit 6cef1e5b authored by Sebastian Schüpbach's avatar Sebastian Schüpbach
Browse files

use mariadb prepared statements

parent 6ce34283
Pipeline #19196 failed with stages
in 41 seconds
import logging
import numbers
import os
import time
import MySQLdb
import mysql.connector as mariadb
......@@ -39,36 +37,31 @@ class Indexer:
def _create_sql_stmt(table_name, record, fields) -> str:
def _create_sql_stmt(table_name, record, fields) -> (str, tuple):
Create SQL statement
db_fields = [dbField for dbField in fields
if dbField in record and record[dbField] is not None]
db_values = [str(record[db_field])
if isinstance(record[db_field], numbers.Number)
else MySQLdb.escape_string(record[db_field])
for db_field in db_fields]
key_value = \
", ".join([k + "=" + v for (k, v) in zip(db_fields, db_values) if k != 'sig'])
db_fields = ','.join(db_fields)
db_values = ','.join(db_values)
db_values = [record.get(f) for f in fields]
db_values.extend([record.get(f) for f in fields if f != 'sig'])
db_fields = ','.join(fields)
db_value_placeholders = ', '.join(['?' for _ in fields])
key_value = ", ".join([f"{f}=?" for f in fields if f != 'sig'])
# noinspection SqlNoDataSourceInspection
return 'INSERT INTO {} ({}) VALUES ({}) ON DUPLICATE KEY UPDATE {}'.format(
table_name, db_fields, db_values, key_value)
table_name, db_fields, db_value_placeholders, key_value), tuple(db_values)
def insert_in_db(self, record) -> (bool, str):
Insert record in DB
entities_stmt = Indexer._create_sql_stmt('entities', record,
entities_stmt, entities_values = Indexer._create_sql_stmt('entities', record,
['sig', 'uri', 'access', 'proto'])
metadata_stmt = Indexer._create_sql_stmt('metadata', record,
metadata_stmt, metadata_values = Indexer._create_sql_stmt('metadata', record,
['sig', 'mimetype', 'height',
'width', 'duration', 'type'])
self.mariadb_cursor.execute(entities_stmt, entities_values)
self.mariadb_cursor.execute(metadata_stmt, metadata_values)
return True, ""
except mariadb.Error as ex:
logging.error("Problems in sql statement: {}".format(entities_stmt))
# mediametadatatodb
# Copyright (C) 2020 Memoriav
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
from mediametadatatodb_app.resources import indexer
class Test(unittest.TestCase):
# noinspection SqlNoDataSourceInspection,SqlResolve
def test__create_sql_stmt(self):
record = {
'sig': 'test-001',
'mimetype': 'image/jpeg',
'height': 20,
'width': 100,
'type': 'image'
metadata_stmt, metadata_values = \
indexer.Indexer._create_sql_stmt('metadata', record,
['sig', 'mimetype', 'height',
'width', 'duration', 'type'])
self.assertEqual(("INSERT INTO metadata (sig,mimetype,height,width,duration,type) VALUES"
" (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE mimetype=?, height=?,"
" width=?, duration=?, type=?",
('test-001', 'image/jpeg', 20, 100, None, 'image',
'image/jpeg', 20, 100, None, 'image')),
(metadata_stmt, metadata_values))
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment