GVSDS/NSQL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

The Fisrt Updated

Browse Source
This commit is contained in:
TermiNexus
2026-01-25 12:53:16 +08:00
commit 2b6a1bbe18
3 changed files with 532 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
NSQL.py Normal file
View File

@@ -0,0 +1,226 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
NSQL - Lightweight MySQL Wrapper for Python
Project: MySQL-Wrapper
Author: WYT
Created: 2025/4/20
Version: 0.1.497
GitHub: https://github.com/GVSADS/NSQL/
Copyright (c) 2023 [WYT/GVSDS]. All rights reserved.
License: MIT License
Description:
A thread-safe Python wrapper for PyMySQL with enhanced features
including SQL injection protection, automatic type conversion
and debug mode support.
Dependencies:
- PyMySQL >= 1.0.0
- Python >= 3.6
"""
import threading
import pymysql
import binascii
import traceback
import json
import re
class Light():
def __init__(self):
self.conn = pymysql.Connection
class Disposes():
def __init__(self):
pass
def _PK(self, a=(None), k={None}):
_ = ""
for i in a:
_ += "%s," % self._PS(i)
_ = _[:-1]
return _
def _PS(self, s):
if isinstance(s, _Func):
return str(s)
elif type(s) == str:
return "'%s'" % pymysql.converters.escape_string(s)
elif type(s) == bytes:
return Func.insertbytes(s)
else:
return "%s" % s
def _PR(self, _, head):
if _ == None:
return {}
return _ if head[1] == "text" else int(_) if head[1] == "int" else float(_) if head[1] == "float" else json.loads(_) if head[1] == "json" else _
Disposes = Disposes()
class MySQL():
def __init__(self, host, port, charset="utf8", debug=False):
self.host, self.port, self.charset, self.debug = host, port, charset, debug
self.Lock = threading.Lock()
def __login__(self, user, passwd):
self.conn = pymysql.connect(host=self.host, port=self.port, user=user, passwd=passwd, charset=self.charset)
self.conn.connect_timeout = 1000000000000
class NewCursor():
def __init__(self, parent: Light) -> None:
self.parent = parent
self.conn = self.parent.conn
self.cursor = self.parent.conn.cursor()
self.debug = self.parent.debug
self.Table = None
self.db = None
def to_connect(self):
print("Server DisConnect")
def __run_sql__(self, sql, args=()):
try:
if self.debug:
print("SQL [%s] [%s]: %s" % (self.db, self.Table, sql))
print("Params:", args)
self.parent.Lock.acquire()
if self.db: self.cursor.execute("USE `%s`" % self.db)
self.cursor.execute(sql, args)
self.conn.commit()
return self.cursor.fetchall()
except Exception as e:
if isinstance(e, pymysql.err.InterfaceError):
self.conn = self.to_connect()
print(traceback.format_exc())
return False
if self.conn and self.debug:
print(traceback.format_exc())
return False
finally:
self.parent.Lock.release()
def is_valid_identifier(self, identifier):
return re.match(r"^[a-zA-Z0-9_]+$", identifier) is not None
def use(self, db, Table=None):
if not self.is_valid_identifier(db):
raise ValueError("Invalid database name")
self.Table = Table
self.db = db
return self.__run_sql__("USE `%s`" % db)
def __ChanceFrom__(self, FROM):
return FROM if FROM else self.Table
def __add_fin__(self, FROM=None, WHERE=None, _limit=None):
FROM = self.__ChanceFrom__(FROM)
from_clause = f"FROM `{FROM}`" if FROM else ""
where_clause = ""
where_params = ()
if WHERE is not None:
if isinstance(WHERE, tuple):
if len(WHERE) != 2:
raise ValueError("WHERE IS DICT: (condition_template, params)")
where_cond, where_params = WHERE[0], WHERE[1]
where_clause = f"WHERE {where_cond}"
elif isinstance(WHERE, str):
where_clause = f"WHERE {WHERE}"
else:
raise TypeError("WHERE BE MUST DICT OR VALUE")
limit_clause = f"LIMIT {_limit}" if _limit else ""
if not isinstance(where_params, tuple):
where_params=(where_params,)
return (
f"{from_clause} {where_clause} {limit_clause}".strip(),
where_params
)
def __load_sql__(self, run, _table, FROM=None, WHERE=None, _limit=None):
clauses, where_params = self.__add_fin__(FROM, WHERE, _limit)
full_sql = f"{run} {_table} {clauses}"
return self.__run_sql__(full_sql, where_params)
def delete(self, FROM=None, WHERE=None):
return self.__load_sql__("DELETE", "", FROM, WHERE)
def insert(self, _Table, values=None, WHERE=None, **k):
if values is None:
values = k
elif k:
raise ValueError("CAN'T SET VALUE AND KEY AT SAME TIME")
columns = []
params = []
for key, value in values.items():
if not self.is_valid_identifier(key):
raise ValueError(f"EKEY: {key}")
columns.append(key)
params.append(value)
columns_str = ", ".join([f"`{col}`" for col in columns])
placeholders = ", ".join(["%s" if not isinstance(v, _Func) else str(v) for v in params])
clauses, clause_params = self.__add_fin__(WHERE=WHERE)
sql = f"INSERT INTO `{_Table}` ({columns_str}) VALUES ({placeholders}) {clauses}"
return self.__run_sql__(sql, tuple(v for v in params if not isinstance(v, _Func)) + clause_params)
def show(self, _Table, FROM=None, WHERE=None):
return self.__load_sql__("SHOW", _Table, FROM, WHERE)
def select(self, _Table, FROM=None, WHERE=None, _limit=None):
if (res := self.__load_sql__("SELECT", _Table, FROM, WHERE, _limit)):
head=self.show(MOD.COLUMNS,FROM=FROM)
tables=[]
heads=[]
for _ in head:
heads.append(_[0])
if _Table not in heads:
for i in res:
mr=[]
for s, _ in enumerate(i):
mr.append(Disposes._PR(_,head[s]))
tables.append(mr)
return tables
else:
for i in head:
if i[0] == _Table:
return Disposes._PR(res[0][0],i)
return False
return res
def selectashead(self, _Table, FROM=None, WHERE=None, _limit=None):
if (res := self.__load_sql__("select", _Table, FROM, WHERE, _limit)):
head = self.show(MOD.COLUMNS, FROM=FROM)
tables = []
for i in res:
mr = {}
for s, _ in enumerate(i):
mr[head[s][0]] = Disposes._PR(_, head[s])
tables.append(mr)
return tables
return res
def update(self, WHERE, FROM=None, **k):
set_fields = []
set_params = []
for key, value in k.items():
set_fields.append(f"{key}=%s")
set_params.append(value)
where_clause, where_params = self.__add_fin__(WHERE=WHERE)
all_params = tuple(set_params) + where_params
sql = f"UPDATE `{FROM}` SET {', '.join(set_fields)} {where_clause}"
return self.__run_sql__(sql, all_params)
def insertasdict(self, _dict, TableName, WHERE=None):
self.insert(TableName, values=_dict, WHERE=WHERE)
def istrue(self, FROM=None, WHERE=None):
return bool(self.select(1, FROM=FROM, WHERE=WHERE, _limit=1))
class MOD():
ALL = "*"
FROM = "FROM"
WHERE = "WHERE"
CREATE = "CREATE"
TABLE = "TABLE"
COLUMNS = "COLUMNS"
class _Func(str):
def __init__(self, s):
str.__init__(self)
class Func():
@staticmethod
def NOW():
return _Func("NOW()")
@staticmethod
def JSON_ARRAY(*a):
return _Func("JSON_ARRAY(%s)" % Disposes._PK(a))
@staticmethod
def LOAD_FILE(path):
return _Func("LOAD_FILE('%s')" % path)
@staticmethod
def UNHEX(s):
return _Func("UNHEX('%s')" % s)
@staticmethod
def insertbytes(s, charset: str = ""):
return _Func("0x%s" % binascii.hexlify(s).decode() if type(s) == bytes else binascii.hexlify(str(s).encode(charset)).decode())
@staticmethod
def insert16(s):
return _Func("0x%s" % s)

305
README.MD Normal file
View File

@@ -0,0 +1,305 @@
# MySQL Wrapper Project - New SQL (NSQL)
A lightweight, thread-safe Python wrapper for PyMySQL with enhanced features.
![Python](https://img.shields.io/badge/Python-3.6+-blue.svg)
![MySQL](https://img.shields.io/badge/MySQL-5.7+-orange.svg)
![PyMySQL](https://img.shields.io/badge/PyMySQL-1.0+-green.svg)
## Table of Contents
- [Advantages](#advantages)
- [Technical Principles](#technical-principles)
- [Features](#features)
- [API Reference](#api-reference)
- [Usage Examples](#usage-examples)
- [Limitations](#limitations)
- [Comparison](#comparison-with-other-projects)
## Advantages
### Compared to Other MySQL Wrappers
✔ **Thread Safety** - Built-in threading lock mechanism
✔ **SQL Injection Protection** - Strict identifier validation
✔ **Automatic Type Conversion** - Smart result type handling
✔ **Flexible Parameter Binding** - Supports both tuple and dict parameters
✔ **Connection Resilience** - Automatic reconnection handling
✔ **Debug Mode** - Detailed SQL logging for development
## Technical Principles
### Transaction Handling
- Uses PyMySQL's native transaction support
- Automatic `COMMIT` after each successful operation
- Manual transaction control available through raw connection
### Connection Pool
- Not a traditional connection pool
- Single persistent connection with thread locking
- Lightweight alternative for moderate workloads
- Suitable for long-running applications
## Features
### Core Features
- Parameterized query building
- Automatic FROM clause completion
- JSON data type support
- Binary data handling utilities
- Debug mode with SQL logging
- Dictionary-style result formatting
### Security Features
- SQL injection prevention
- Strict identifier validation
- Proper string escaping
- Parameter separation from queries
## API Reference
### Main Classes
#### `MySQL(host, port, charset="utf8", debug=False)`
Main wrapper class constructor.
#### `NewCursor(parent)`
Cursor class with enhanced methods.
### Core Methods
| Method | Description | Parameters |
|--------|-------------|------------|
| `use(db, Table=None)` | Switch database | `db`: database name |
| `select(_Table, FROM=None, WHERE=None, _limit=None)` | Basic SELECT | `_Table`: columns to select |
| `selectashead(_Table, FROM=None, WHERE=None, _limit=None)` | Dict-style results | Same as select |
| `insert(_Table, values=None, WHERE=None, **k)` | INSERT operation | Supports dict or kwargs |
| `update(WHERE, FROM=None, **k)` | UPDATE operation | WHERE clause required |
| `delete(FROM=None, WHERE=None)` | DELETE operation | |
| `istrue(FROM=None, WHERE=None)` | Existence check | Returns boolean |
### Helper Functions
| Function | Description | Example |
|----------|-------------|---------|
| `Func.NOW()` | Current timestamp | `Func.NOW()` |
| `Func.JSON_ARRAY()` | JSON array builder | `Func.JSON_ARRAY(1,2,3)` |
| `Func.insertbytes()` | Binary data handler | `Func.insertbytes(b'data')` |
## Usage Examples
### Basic Usage
```python
db = MySQL('localhost', 3306, debug=True)
db.__login__('user', 'password')
cursor = db.NewCursor()
# Select example
cursor.use('mydb', 'users')
results = cursor.select('*', WHERE=('age > %s', (18,)), _limit=10)
# Insert example
cursor.insert('users', {'name': 'John', 'age': 25})
# Transaction example
try:
cursor.update(WHERE=('id=%s', (1,)), FROM='users', balance=100)
cursor.update(WHERE=('id=%s', (2,)), FROM='users', balance=200)
except:
cursor.conn.rollback()
```
### Advanced Features
```python
# Binary data insertion
cursor.insert('files', {
'name': 'data.bin',
'content': Func.insertbytes(b'\x00\x01\x02')
})
# JSON data handling
cursor.insert('config', {
'settings': Func.JSON_ARRAY('item1', 'item2')
})
# Dictionary-style results
users = cursor.selectashead('*', FROM='users')
for user in users:
print(user['name'], user['age'])
```
## Limitations
### Not Recommended For
❌ High-concurrency applications (consider connection pool)
❌ Complex transaction scenarios
❌ ORM-like object mapping
❌ Asynchronous applications
### Performance Considerations
- Single connection model may bottleneck under heavy load
- Not optimized for bulk operations
- Type conversion adds minor overhead
## Comparison with Other Projects
| Feature | This Wrapper | PyMySQL | SQLAlchemy | Django ORM |
|---------|-------------|---------|------------|------------|
| Thread Safety | ✔ | ✖ | ✔ | ✔ |
| Connection Pool | ✖ | ✖ | ✔ | ✔ |
| ORM Features | ✖ | ✖ | ✔ | ✔ |
| SQL Building | ✔ | ✖ | ✔ | ✔ |
| Binary Support | ✔ | ✔ | ✔ | ✔ |
| Debug Mode | ✔ | ✖ | Partial | Partial |
---
# MySQL 封装项目 - New SQL (NSQL)
一个轻量级、线程安全的 PyMySQL Python 封装器，具有增强功能。
![Python](https://img.shields.io/badge/Python-3.6+-blue.svg)
![MySQL](https://img.shields.io/badge/MySQL-5.7+-orange.svg)
![PyMySQL](https://img.shields.io/badge/PyMySQL-1.0+-green.svg)
## 目录
- [优势](#优势)
- [技术原理](#技术原理)
- [功能特性](#功能特性)
- [API参考](#api参考)
- [使用示例](#使用示例)
- [局限性](#局限性)
- [对比](#与其他项目的对比)
## 优势
### 与其他MySQL封装器的比较
✔ **线程安全** - 内置线程锁机制
✔ **SQL注入防护** - 严格的标识符验证
✔ **自动类型转换** - 智能结果类型处理
✔ **灵活参数绑定** - 支持元组和字典参数
✔ **连接弹性** - 自动重连处理
✔ **调试模式** - 详细的SQL日志记录
## 技术原理
### 事务处理
- 使用PyMySQL原生事务支持
- 每次成功操作后自动提交`COMMIT`
- 可通过原始连接手动控制事务
### 连接池
- 非传统连接池
- 带线程锁的单一持久连接
- 适用于中等工作负载的轻量级方案
- 适合长期运行的应用程序
## 功能特性
### 核心功能
- 参数化查询构建
- 自动FROM子句补全
- JSON数据类型支持
- 二进制数据处理工具
- 带SQL日志记录的调试模式
- 字典式结果格式化
### 安全特性
- SQL注入预防
- 严格标识符验证
- 正确的字符串转义
- 查询与参数分离
## API参考
### 主要类
#### `MySQL(host, port, charset="utf8", debug=False)`
主封装类构造函数
#### `NewCursor(parent)`
带增强方法的游标类
### 核心方法
| 方法 | 描述 | 参数 |
|------|------|------|
| `use(db, Table=None)` | 切换数据库 | `db`: 数据库名 |
| `select(_Table, FROM=None, WHERE=None, _limit=None)` | 基础SELECT | `_Table`: 选择列 |
| `selectashead(_Table, FROM=None, WHERE=None, _limit=None)` | 字典式结果 | 同select |
| `insert(_Table, values=None, WHERE=None, **k)` | INSERT操作 | 支持字典或关键字参数 |
| `update(WHERE, FROM=None, **k)` | UPDATE操作 | 必须包含WHERE子句 |
| `delete(FROM=None, WHERE=None)` | DELETE操作 | |
| `istrue(FROM=None, WHERE=None)` | 存在性检查 | 返回布尔值 |
### 辅助函数
| 函数 | 描述 | 示例 |
|------|------|------|
| `Func.NOW()` | 当前时间戳 | `Func.NOW()` |
| `Func.JSON_ARRAY()` | JSON数组构建器 | `Func.JSON_ARRAY(1,2,3)` |
| `Func.insertbytes()` | 二进制数据处理 | `Func.insertbytes(b'data')` |
## 使用示例
### 基础用法
```python
db = MySQL('localhost', 3306, debug=True)
db.__login__('user', 'password')
cursor = db.NewCursor()
# 查询示例
cursor.use('mydb', 'users')
results = cursor.select('*', WHERE=('age > %s', (18,)), _limit=10)
# 插入示例
cursor.insert('users', {'name': '张三', 'age': 25})
# 事务示例
try:
cursor.update(WHERE=('id=%s', (1,)), FROM='users', balance=100)
cursor.update(WHERE=('id=%s', (2,)), FROM='users', balance=200)
except:
cursor.conn.rollback()
```
### 高级功能
```python
# 二进制数据插入
cursor.insert('files', {
'name': 'data.bin',
'content': Func.insertbytes(b'\x00\x01\x02')
})
# JSON数据处理
cursor.insert('config', {
'settings': Func.JSON_ARRAY('item1', 'item2')
})
# 字典式结果
users = cursor.selectashead('*', FROM='users')
for user in users:
print(user['name'], user['age'])
```
## 局限性
### 不推荐场景
❌ 高并发应用（考虑连接池方案）
❌ 复杂事务场景
❌ 类ORM的对象映射
❌ 异步应用
### 性能考量
- 单连接模型在重负载下可能成为瓶颈
- 未针对批量操作优化
- 类型转换会增加少量开销
## 与其他项目的对比
| 特性 | NSQL | PyMySQL | SQLAlchemy | Django ORM |
|------|---------|---------|------------|------------|
| 线程安全 | ✔ | ✖ | ✔ | ✔ |
| 连接池 | ✖ | ✖ | ✔ | ✔ |
| ORM功能 | ✖ | ✖ | ✔ | ✔ |
| SQL构建 | ✔ | ✖ | ✔ | ✔ |
| 二进制支持 | ✔ | ✔ | ✔ | ✔ |
| 调试模式 | ✔ | ✖ | 部分 | 部分 |

1
requirements.txt Normal file
View File

@@ -0,0 +1 @@
PyMySQL==1.1.1