现代的编程语言在进行数据库操作时候，往往都是使用ORM组件（Object Relational Mapping）而不是直接链接到数据库中然后自己写SQL，ORM可以帮我们厘清数据关系、自动进行内置的Sql注入防护等等，总之好处多多。但是往往有点不太自由束手束脚不太放得开。
在早期的PHP写Web应用时候基本都是自己拼Sql语句的，当然带来大量的注入式漏洞。时下在Golang语言中，其语言内置了数据库的Sql包，可以让用户直接通过非ORM、非查询构建器方法来与数据库交互。即使在一些Golang大型项目中，写Sql语句也是很常见的。
而现在最流行，用户量最大的Python语言中没有任何内置的支持数据库交互的东西，这也给初学者进行数据库操作增加麻烦。本文中虫虫就给大家介绍如何在Python中通过适配器直接编写Sql语句，从而达到间接学习数据库和Sql语言的目的。
本文主要目的在于学习，可以实现在Python中不依赖查询生成器和ORM熟练的拼写和处理Sql，并能将这些逻辑抽象打包，最后可以实现MVC三层架构，实现数据库操作和业务逻辑（以及表现层）分开。
从一个简单的例子开始：
# user/domain.py
@dataclass
class User:
id: int = None
dt_created: datetime = None
username: str = None
email: str = None
mobile: Optional[str] = None
User是可能看到的名为SQL表的字面定义user。 这包括喜欢包含在所有SQL表中的一些列：
id（主键）。
dt_created日期时间值，通常默认为NOW()。
所有字段默认为None，这可以姐可以执行SELECT不包含表中所有列的查询，同时仍然能够将行转换为User对象。所有列都是NOT NULL除了mobile，代码中利用Optional键入提示以强调该字段允许NULL。
创建数据库
# user/repository.py
from abc import ABC, abstractmethod
class UserRepository(ABC):
@abstractmethod
def new(self, new_row: User) -> int:
'''
Create a new record and returns the new id value.
'''
pass
@abstractmethod
def get_by_id(self, id_val: int) -> User:
'''
Get a single record by ID value.
'''
pass
现在可以创建的存储库，创建的存储库是一个抽象类，可以确保任何未来的实现都需要遵循规则。
为了方便，我们直接使用psycopg3来实现PostgreSQL链接，其他数据库类型都可以随时扩展：
# external/postgres.py
import psycopg
from psycopg.rows import class_row
from config import get_config
cfg = get_config()
conn = psycopg.connect(
dbname=cfg.database_name,
user=cfg.database_username,
password=cfg.database_password,
host=cfg.database_host,
)
def new_cursor(name=None, row_factory=None):
if name is None:
name = ''
if row_factory is None:
return conn.cursor(name=name)
return conn.cursor(name=name, row_factory=class_row(row_factory))
首先，根据配置设置连接，连接到数据。
结果会返回psycopg游标的函数。psycopg3包含一个自定义行工厂，这个功能非常有用。
# user/repository.py
from external.postgres import conn, new_cursor
class UserPostgreSQL(UserRepository):
def new(self, new_row: User) -> int:
with new_cursor() as cur:
cur.execute(
'''
INSERT INTO user
(username, email)
VALUES
(%s, %s)
ON CONFLICT (email)
DO NOTHING
RETURNING id;
''',
(
new_row.username,
new_row.email
),
)
new_id = cur.fetchone()
if new_id and len(new_id):
conn.commit()
new_id = new_id[0]
else:
new_id = 0
return new_id
def get_by_id(self, id_val: int) -> User:
with new_cursor(name='get_by_id', row_factory=User) as cur:
cur.execute(
'''
SELECT *
FROM user
WHERE id = %s
''',
(id_val,),
)
return cur.fetchone()
def new_user_repo() -> UserRepository:
return UserPostgreSQL()
你会注意到new()使用基本光标，但是get_by_id()游标使用带有类行工厂的 ，这这样返回的行将是类型User对象。
new_user_repo()在假设的业务逻辑中使用的函数：
from user.repository import *
user_repo = new_user_repo()
row = User()
row.username = 'chognchong'
row.email = 'chongchong(at)someadress'
# Create a new record.
new_id = user_repo.new(row)
# Fetch the record we just created.
new_row = user_repo.get_by_id(new_id)
# User(id=1, dt_created=datetime.datetime(2023, 8, 9, 23, 8, 14, 974074, tzinfo=datetime.timezone.utc), username=' chognchong', email=' chongchong(at)someadress ', mobile=None)
new_user_repo()可以修改为返回任何实现UserRepository，可以是PostgreSQL，也可能是MySQL实现，或者SQLite实现，甚至可能是用于测试目的的模拟数据库。
总结
在编程届各种技术层出不穷，迭代替换，唯一常青树不变的，SQL是一个。编写Sql可以让我们更加轻松自由和数据库打交道，当然对于初学者更是莫大的宝藏，与其看看几个月就会被淘汰的ORM复杂语法框架，不如直接深入纯的Sql层。当然直接写Sql一直有风险，尤其是线上系统，请谨慎使用。
联系客服
