preface
The main content of this article is to use Python to write some practical scripts to facilitate our daily use (as long as we dare to think, we dare to create).
In daily life, we write the front-end and back-end, and often use the database. MySQL is often used in Python to operate mysql. Usually, we need to write a lot of sql code and prevent sql injection. We can use a simple addition, deletion, modification and query statement generation function to help us generate statements quickly and reduce unnecessary time-consuming; Moreover, most entity classes need to be defined to help improve the model. For example, when using fastapi, you need to use the subclass of BaseModel as the model class to judge the type and define the data format. If each entity class needs to write sql statements separately, it may be troublesome to check whether there are databases and data tables. Establishing a unified class can reduce our daily useless development. Of course, sql is simpler and supports addition, deletion and query. Those sub queries and left to right connections can use multiple queries in the face of small traffic and small access, It will be easier for Python. For complex queries, it is still implemented through complete sql statements.
Database help class
This idea came into being when I saw that there was a DBHelper class in Java when I actually used django to automatically create databases and data tables. I feel good. Share it.
Information setting function
The basic connection information can be set here. Since Python imports the file in singleton mode, it will be set once, and the set variables will be used throughout the whole life cycle of the program.
MYSQL_HOST = 'localhost'
MYSQL_PORT = 3306
MYSQL_USER = 'root'
MYSQL_PASSWORD = ''
MYSQL_DATABASE = ''
RESTART_INIT = True
mysql = ...
cursor = ...
def setting(host='localhost', port=3306, user='root', password='password', database='database', init=True):
"""
Set global variables
:param host: Database host name(address)
:param port: Database port
:param user: Database user
:param password: Database user password
:param database: Database name
:param init: Does the program initialize the database each time it starts
:return: NoReturn
"""
global MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE, RESTART_INIT, mysql, cursor
MYSQL_HOST = host
MYSQL_PORT = port
MYSQL_USER = user
MYSQL_PASSWORD = password
MYSQL_DATABASE = database
RESTART_INIT = init
try:
mysql = pymysql.connect(
host=MYSQL_HOST,
port=MYSQL_PORT,
user=MYSQL_USER,
password=MYSQL_PASSWORD,
charset='utf8mb4',
cursorclass=pymysql.cursors.SSDictCursor
)
cursor = mysql.cursor()
if RESTART_INIT:
cursor.execute(f'DROP DATABASE IF EXISTS `{MYSQL_DATABASE}`;')
cursor.execute(f'CREATE DATABASE IF NOT EXISTS `{MYSQL_DATABASE}`;')
mysql.commit()
cursor.close()
mysql.close()
mysql = pymysql.connect(
host=MYSQL_HOST,
port=MYSQL_PORT,
user=MYSQL_USER,
password=MYSQL_PASSWORD,
database=MYSQL_DATABASE,
charset='utf8mb4',
cursorclass=pymysql.cursors.SSDictCursor
)
cursor = mysql.cursor()
except pymysql.err.OperationalError:
print('User name and password do not match!')
logging.exception(f'user name {MYSQL_USER} And password {MYSQL_PASSWORD} No match!')
time.sleep(.01)
exit('Although the program exited abnormally, don't panic. Everything is under my control!')
except pymysql.Error as e:
print(f'exception: {e.args}')
logging.exception(e.args)
exit('Although the program exited abnormally, don't panic. Everything is under my control!')
sql query function
Since the query operation does not need to be applied compared with the addition, deletion and modification operation, a function should be placed separately to return the queried content.
def exec_sql(sql):
if mysql is ... or cursor is ...:
print('Please initialize the settings first setting Function!')
logging.error('Please initialize the settings first setting Function!')
exit('Although the program exited abnormally, don't panic. Everything is under my control!')
mysql.ping()
cursor.execute(sql)
res = cursor.fetchall()
if res:
return list(res)
return []
sql execution function
Then, other operations only return whether the execution is successful or not to give different responses.
def commit_sql(sql):
if mysql is ... or cursor is ...:
print('Please initialize the settings first setting Function!')
logging.error('Please initialize the settings first setting Function!')
exit('Although the program exited abnormally, don't panic. Everything is under my control!')
try:
mysql.ping()
cursor.execute(sql)
mysql.commit()
return True
except pymysql.Error as e:
print(f'exception: {e.args}')
logging.exception(e.args)
mysql.rollback()
return False
Integer field
Through the relevant default parameters, the sql statement is automatically generated, and there is a self increasing attribute in the integer.
class Int:
def __init__(self, *, max_length=11, null=False, primary=False, only=False, auto=False, default=''):
sql = 'int'
if max_length:
sql += f'({max_length})'
if null:
sql += ' NOT NULL'
if primary:
sql += ' PRIMARY KEY'
if only:
sql += ' UNIQUE'
if auto:
sql += ' AUTO_INCREMENT'
if default:
sql += f' DEFAULT {default}'
self.sql = sql
Floating point field
Automatically generate sql statements through relevant default parameters.
class Float:
def __init__(self, *, max_length=0, null=False, primary=False, only=False, default=''):
sql = 'float'
if max_length:
sql += f'({max_length})'
if null:
sql += ' NOT NULL'
if primary:
sql += ' PRIMARY KEY'
if only:
sql += ' UNIQUE'
if default:
sql += f' DEFAULT {default}'
self.sql = sql
String field
Automatically generate sql statements through relevant default parameters.
class Varchar:
def __init__(self, *, max_length=255, null=False, primary=False, only=False, default=''):
sql = 'varchar'
if max_length:
sql += f'({max_length})'
sql += ' CHARACTER SET utf8mb4'
if null:
sql += ' NOT NULL'
if primary:
sql += ' PRIMARY KEY'
if only:
sql += ' UNIQUE'
if default:
sql += f' DEFAULT {default}'
self.sql = sql
Text field
Through the relevant default parameters, sql statements are automatically generated. The text type is the scheme when the string length exceeds 255, including short, long, etc. here, the text type is simply used.
class Text:
def __init__(self, *, max_length=0, null=False, primary=False, only=False, default=''):
sql = 'text'
if max_length:
sql += f'({max_length})'
sql += ' CHARACTER SET utf8mb4'
if null:
sql += ' NOT NULL'
if primary:
sql += ' PRIMARY KEY'
if only:
sql += ' UNIQUE'
if default:
sql += f' DEFAULT {default}'
self.sql = sql
Data table class
This is a class that automatically creates a data table according to the field during initialization and automatically performs simple addition, deletion, modification and query on the data table according to the conditions.
class Table:
def __init__(self, table, **kwargs):
sql = f'CREATE TABLE IF NOT EXISTS `{table}` (' + ', '.join([f'`{key}` {kwargs[key].sql}' for key in kwargs]) + ') DEFAULT CHARSET=utf8mb4;'
self.fields = kwargs
self.table = table
self.sql = sql
self.create()
def field(self):
return list(self.fields.keys())
def record(self):
res = self.select(fields='COUNT(*)')
if res:
return res[0]['COUNT(*)']
return 0
def create(self):
if commit_sql(self.sql):
return True
else:
return False
def select(self, *, fields='*', where=None):
if isinstance(fields, list):
fields = ', '.join(fields)
sql = f'SELECT {fields} FROM `{self.table}`'
if where:
sql += ' WHERE ' + ' AND '.join([f'`{key}` = "{where[key]}"' for key in where])
sql += ';'
return exec_sql(sql)
def insert(self, *, fields=False, values):
sql = f'INSERT INTO `{self.table}` '
if fields:
sql += '(' + ', '.join([f'`{key}`' for key in values.keys()]) + ') VALUES (' + ', '.join([f'"{value}"' for value in values.values()]) + ');'
elif isinstance(values, dict):
sql += 'VALUES (' + ', '.join([f'"{value}"' for value in values.values()]) + ');'
elif isinstance(values, list):
sql += 'VALUES (' + ', '.join([f'"{value}"' for value in values]) + ');'
else:
sql += 'VALUES (' + str(values) + ');'
if commit_sql(sql):
return True
else:
return False
def update(self, *, where, values):
sql = f'UPDATE `{self.table}` SET ' + ', '.join([f'`{key}` = "{values[key]}"' for key in values]) + ' WHERE ' + ' AND '.join([f'`{key}` = "{where[key]}"' for key in where]) + ';'
if commit_sql(sql):
return True
else:
return False
def delete(self, *, where):
sql = f'DELETE FROM `{self.table}` WHERE ' + ' AND '.join([f'`{key}` = "{where[key]}"' for key in where]) + ';'
if commit_sql(sql):
return True
else:
return False
Entity generation class
This class is used to quickly generate entity classes, with type check and regular check when initializing classes, which can effectively prevent sql injection.
property function generation function
Usually, this function can check the type during assignment. If there are many attributes, it will be written similarly. I designed a generation function. As soon as the relevant parameters are input, the output of direct assignment can be pasted.
def make_property_function(variable_list, method_list, *, indentation=1, condition=True, private='__'):
"""
generate property Function( \t Not be pycharm Finally, you need to \t Convert to 4 spaces)
:param variable_list: String type description
:param method_list: property Method list ['getter', 'setter', 'deleter'] (set up property Function, property This experience is generated by default)
:param indentation: Function indentation level refers to the indentation before the line where the decorator and function name are located (one for each level) \t Length)
:param condition: Whether to add simple type judgment (when variable_list Cannot use this feature when is a list)
:param private: Private hierarchy ['_', '__']
:return: Generated property Function string
"""
if isinstance(variable_list, list):
condition = False
string = ''
for v in variable_list:
string += '\n\n' + '\t' * indentation + '@property'
string += '\n' + '\t' * indentation + f'def {v}(self):'
string += '\n' + '\t' * indentation + f'\treturn self.{private}{v}'
builtins = ''
if v in dir(__builtins__):
builtins = '_'
if 'setter' in method_list:
string += '\n\n' + '\t' * indentation + f'@{v}.setter'
string += '\n' + '\t' * indentation + f'def {v}(self, {builtins}{v}):'
if condition:
string += '\n' + '\t' * indentation + f'\tif not isinstance({builtins}{v}, {variable_list[v]}):'
string += '\n' + '\t' * indentation + f'\t\traise ValueError("variable `{v}` must be type of `{variable_list[v]}`")'
string += '\n' + '\t' * indentation + f'\tself.{private}{v} = {builtins}{v}'
if 'getter' in method_list:
string += '\n\n' + '\t' * indentation + f'@{v}.getter'
string += '\n' + '\t' * indentation + f'def {v}(self):'
string += '\n' + '\t' * indentation + f'\treturn self.{private}{v}'
if 'deleter' in method_list:
string += '\n\n' + '\t' * indentation + f'@{v}.deleter'
string += '\n' + '\t' * indentation + f'def {v}(self):'
string += '\n' + '\t' * indentation + f'\tdel self.{private}{v}'
return string.replace('\t', ' ')
property function generating function stub generating function
This is the stub generation function of the function. In order to be forced, I usually write a stub file, but handwriting is also very tired. Therefore, this function is designed to use unified parameters with the function generation function.
def make_property_stub(variable_list, method_list, *, indentation=1):
"""
generate property Stub string of function
:param variable_list: String type description
:param method_list: property Method list ['getter', 'setter', 'deleter'] (set up property Function, property This experience is generated by default)
:param indentation: Function indentation level refers to the indentation before the line where the decorator and function name are located (one for each level) \t Length)
:return:
"""
string = ''
for v in variable_list:
string += '\n' + '\t' * indentation + '@property'
string += '\n' + '\t' * indentation + f'def {v}(self) -> {variable_list[v]}: ...'
builtins = ''
if v in dir(__builtins__):
builtins = '_'
if 'setter' in method_list:
string += '\n' + '\t' * indentation + f'@{v}.setter'
string += '\n' + '\t' * indentation + f'def {v}(self, {builtins}{v}: {variable_list[v]}) -> NoReturn: ...'
if 'getter' in method_list:
string += '\n' + '\t' * indentation + f'@{v}.getter'
string += '\n' + '\t' * indentation + f'def {v}(self) -> {variable_list[v]}: ...'
if 'deleter' in method_list:
string += '\n' + '\t' * indentation + f'@{v}.deleter'
string += '\n' + '\t' * indentation + f'def {v}(self) -> NoReturn: ...'
return string.replace('\t', ' ')
Functions for generating Python code files and python stub files
This is used in the following way: make_python_file('class name ', parameter type, required parameter, non required parameter) other parameters are optional, and often only MySQL needs to be changed_ tools_ The reference address of file (database help class) and the list of method names required by the property function.
It also adds the function of "gift modification" in the same body, which is automatically created when the unique key is not found. Otherwise, you can modify the definition yourself when updating the data. At the same time, it adds the deletion function, which is less used. There's a make_ all_??? The S (table) function can automatically read the table and generate all data into entity classes for iterative return.
NEWLINE = '\n' # Line feed flag
def make_python_file(file_name, data_type, data_necessary, data_unnecessary, *, project_name='',
mysql_tools_file='MySQLController', create_file=True, property_use=True, method_list=None):
"""
Generation function module .py Documents and .pyi stub file
:param file_name: Model name (also file name, without suffix)
:param data_type: Parameter type
:param data_necessary: Required parameters
:param data_unnecessary: Non required parameters
:param project_name: entry name
:param mysql_tools_file: Relative path of database tool class (for example: ToolsCreator.MySQLController)[.Represents the current directory..Represents the upper level directory]
:param create_file: Create file automatically
generate property function
:param property_use: Create property function
:param method_list: property Method list ['getter', 'setter', 'deleter'] (set up property Function, property This experience is generated by default)
:return: (.py Document content, .pyi Document content)
"""
string = ''
string2 = ''
assert set(data_type.values()) | {int, float, str} == {int, float, str}, ValueError('The current version only supports int,float,str Three types!')
if not method_list:
method_list = []
if property_use:
variable_list = {parameter: data_type[parameter].__name__ for parameter in data_type}
string = make_property_function(variable_list, method_list)
string2 = make_property_stub(variable_list, method_list)
file_py = f"""
# _*_ coding:utf-8 _*_
# Project: {project_name}
# FileName: {file_name.lower()}.py
# ComputerUser: 19305
# Day: {time.strftime('%Y/%m/%d', time.localtime(time.time()))}
# Time: {time.strftime('%H:%M', time.localtime(time.time()))}
# IDE: PyCharm
# In 2022, all bug s will be thrown into the sea to feed sharks! I said it! Irrefutable!
import re
from {mysql_tools_file} import setting, Table, Int, Float, Varchar
{file_name.upper()}_TYPE = {{{', '.join(f'{parameter!r}: {data_type[parameter].__name__}' for parameter in data_type)}}}
{file_name.upper()}_NECESSARY = {data_necessary!r}
{file_name.upper()}_UNNECESSARY = {data_unnecessary!r}
{file_name.upper()}_RE_TYPE = {{
{f',{NEWLINE}'.join(f" {parameter!r}: re.compile(R'.*')" for parameter in data_type)}
}}
{file_name.upper()}_FIELD_TYPE = {{
'id': Int(null=True, primary=True, auto=True), # Auto increment id
{f',{NEWLINE}'.join(f' {parameter!r}: {(data_type[parameter] == int and "Int(null=True)") or (data_type[parameter] == float and "Float(null=True)") or "Varchar(null=True)"}' for parameter in data_necessary)}{',' if data_unnecessary else ''}
{f',{NEWLINE}'.join(f' {parameter!r}: {(data_type[parameter] == int and "Int(null=False)") or (data_type[parameter] == float and "Float(null=False)") or "Varchar(null=False)"}' for parameter in data_unnecessary)}{NEWLINE if data_unnecessary else ''}}}
def check(function):
def type_check(self, **kwargs):
if kwargs.get('id'):
self._{file_name.title()}__id = kwargs.get('id')
del kwargs['id']
for n, attributes in enumerate(zip(list(kwargs.keys()), tuple(kwargs.values()))):
if {file_name.upper()}_TYPE.get(attributes[0], str) == float and isinstance(attributes[1], int):
if kwargs.get(attributes[0], None):
kwargs[attributes[0]] = float(kwargs[attributes[0]])
elif {file_name.upper()}_TYPE.get(attributes[0], str) == str:
kwargs[attributes[0]] = str(kwargs[attributes[0]])
elif not isinstance(attributes[1], {file_name.upper()}_TYPE.get(attributes[0], str)):
raise ValueError(f'Variable `{{attributes[0]}}` must be type of `{{{file_name.upper()}_TYPE.get(attributes[0], str)}}`')
elif not {file_name.upper()}_RE_TYPE.get(attributes[0], re.compile('.*?')).fullmatch(str(attributes[1])):
raise ValueError(f'Variable `{{attributes[0]}}`="{{attributes[1]}}" is not a correct value`')
return function(self, **kwargs)
return type_check
class {file_name.title()}:
@check
def __init__(self, *, {', '.join(data_necessary)}{''.join(f', {parameter}=None' for parameter in data_unnecessary)}):
{NEWLINE.join(f' self.__{parameter} = {parameter}' for parameter in data_type)}
def commit(self, table): # Insert or update data
if table.select(fields='{data_necessary[0]}', where=dict({data_necessary[0]}=self.__{data_necessary[0]})):
return table.update(where=dict({data_necessary[0]}=self.__{data_necessary[0]}), values={{key: self.__dict__.get(f'_{file_name.title()}__{{key}}') for key in {file_name.upper()}_NECESSARY + {file_name.upper()}_UNNECESSARY if self.__dict__.get(f'_{file_name.title()}__{{key}}')}})
else:
return table.insert(fields=True, values={{key: self.__dict__.get(f'_{file_name.title()}__{{key}}') for key in {file_name.upper()}_NECESSARY + {file_name.upper()}_UNNECESSARY if self.__dict__.get(f'_{file_name.title()}__{{key}}')}})
def delete(self, table): # Delete data
if table.select(fields='{data_necessary[0]}', where=dict({data_necessary[0]}=self.__{data_necessary[0]})):
return table.delete(where=dict({data_necessary[0]}=self.__{data_necessary[0]}))
return False{string}
def make_all_{file_name.lower()}s(table):
def format_data(data):
data = {{key: {file_name.upper()}_TYPE[key](data[key]) if key != 'id' else int(data[key]) for key in data if data[key]}} # Change the processing method when the key is id
return data
for {file_name.lower()} in table.select():
yield {file_name.title()}(**format_data({file_name.lower()}))
if __name__ == '__main__':
setting()
{file_name.lower()}s = Table('{file_name.lower()}s', **{file_name.upper()}_FIELD_TYPE)
""".lstrip('\n')
file_byi = f"""
from re import Pattern
from types import FunctionType
from {mysql_tools_file} import Int, Float, Varchar, Table
from typing import Iterator, Dict, List, Callable, NoReturn
{file_name.upper()}_TYPE: Dict[str, Callable] = ... # Parameter type
{file_name.upper()}_NECESSARY: List[str] = ... # Required parameters
{file_name.upper()}_UNNECESSARY: List[str] = ... # Non required parameters
{file_name.upper()}_RE_TYPE: Dict[str, Pattern] = ... # Parameter string regularity test
{file_name.upper()}_FIELD_TYPE: Dict[str, Int | Float | Varchar] = ... # Parameter database field correspondence
def check(function) -> FunctionType: # Check function decorator
def type_check(self: Callable = ..., **kwargs: Dict[str, int | float | str]) -> NoReturn | ValueError: ... # Check whether the parameters conform to the type
class {file_name.title()}:
def __init__(self, *,
{f',{NEWLINE}'.join(f' {parameter}: {data_type[parameter].__name__}' for parameter in data_necessary)}{',' if data_unnecessary else ''}
{f',{NEWLINE}'.join(f' {parameter}: {data_type[parameter].__name__} = ...' for parameter in data_unnecessary)}
) -> NoReturn:
{NEWLINE.join(f' self.__{parameter}: {data_type[parameter].__name__} = ...' for parameter in data_type)}
def commit(self, table: Table) -> bool: ...
def delete(self, table: Table) -> bool: ...{string2}
def make_all_{file_name.lower()}s(table: Table) -> Iterator[{file_name.title()}]: # Generate all objects
def format_data(data: [str, int | float | str]) -> Dict[str, int | float | str]: ... # Format data
""".strip('\n')
if create_file:
open(f'{file_name.lower()}.py', 'w', encoding='utf-8').write(file_py)
open(f'{file_name.lower()}.pyi', 'w', encoding='utf-8').write(file_byi)
return file_py, file_byi
Concluding remarks
This is the whole content of my tool class. If you like, you can click three times~~