day3——利用python创立ORM

2023-12-10 10:32
文章标签 python day3 orm 创立

本文主要是介绍day3——利用python创立ORM,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

  • 写在前面
  • 必备知识
    • mysql数据库
      • RDBMS术语
      • 注意
    • sql语法
    • ORM
  • 编写ORM模块
    • 创建连接池
    • 关闭连接池
    • 封装数据库操作指令
      • select
      • executeupdate insert delete
    • 自顶向下的设计方案
      • 字段类
      • Model类
        • 关于元类

写在前面

本文是根据廖雪峰Day 3 - 编写ORM实践后整理的学习笔记,主要记录实践过程中遇到的问题,以及对其所涉及到的知识进行提炼与补充。

必备知识

mysql数据库

数据库(database)即按照数据结构来组织、存储与管理数据的仓库。
mysql数据库是一种关系型数据库(RDBMS),建立在关系模型基础上的数据库,其特点是:

  • 数据以表(table)的形式存储在数据库中(database)
  • 表的每列为记录名称所对应的数据域【可以理解为划分类】
  • 表的每行为记录名称【可以理解为具体属性】

一种常见的数据表

namescoreplace
Mike100China
Jane88US
Mille68UK

RDBMS术语

(这里只罗列一些本文可能涉及到的)

  • 主键(key):主键是唯一的,用来查询数据。即主键的属性值能够唯一定位到一组数据【类似书籍的页码】。
  • 索引:索引可以不唯一,使用索引可以快速访问数据库表中的特定信息【类似书籍的目录】

注意

1.本文所用到的mysql数据库可以去官网下载,但是该数据库只支持python 3.4版本,若要通过python连接数据库,需要下载pymysql模块。
2.本文需要使用到异步aiomysql模块,该模块可能与pymysql模块存在版本不兼容问题。妥善处理方式是,更新aiomysql版本为0.0.7,pymysql版本为0.6.7

sql语法

篇幅有限,这里只解释本项目用到的schema.sql相关语句,具体语法可参考w3school SQL教程。

--schema.sql--如果存在awesome数据库,则删除该数据库(drop)
drop database if exists awesome;  --创建awesome数据库(create database)
create database awesome; --选择awesome数据库(use)
use awesome;--分配权限给特定用户(grant 权限 on 数据库名.表名 to 用户名@登陆方式 identified by 'password')
grant select, insert, update, delete on awesome.* to 'www-data'@'localhost' identified by 'www-data'; --创建users表并设置具体列属性
create table users ( `id` varchar(50) not null,`email` varchar(50) not null,`passwd` varchar(50) not null,`admin` bool not null,`name` varchar(50) not null,`image` varchar(500) not null,`created_at` real not null,key `idx_created_at` (`created_at`), primary key (`id`) 
) engine=innodb default charset=utf8;
--括号内最后两句分别为设置主键,设置索引

执行下列命令即可在mysql数据库中创建相应的数据表。

$ mysql -u root -q < shemal.sql

ORM

说了这么多,这一节是要干嘛?ORM又是什么玩意?

ORM即Object Relational Mapping,全称对象关系映射
当我们需要对数据库进行操作时,势必需要通过连接数据、调用sql语句、执行sql语句等操作,ORM将数据库中的表,字段,行与我们面向对象编程的类及其方法,属性等一一对应,即将该部分操作封装起来,程序猿不需懂得sql语句即可完成对数据库的操作。

编写ORM模块

创建连接池

利用’aiomysql.create_pool()`创建协程连接池连接数据库。

async def create_pool(loop, **kw):'''创建连接池'''logging.info('create database connection pool...')global __pool__pool = await aiomysql.create_pool(host=kw.get('host', 'localhost'),port=kw.get('port', 3306),user=kw['user'],password=kw['password'],db=kw['database'],charset=kw.get('charset', 'utf8'),autocommit=kw.get('autocommit', True),maxsize=kw.get('maxsize', 10),minsize=kw.get('minsize', 1),loop=loop)

注意

数据库的连接打开后一定要及时关闭

关闭连接池

async def close_pool():'''异步关闭连接池'''logging.info('close database connection pool...')global __pool__pool.close()await __pool.wait_closed()

封装数据库操作指令

数据库操作主要包括select, execute(update, insert, delete)。除了select需要返回查询内容,其他命令只需返回一个影响行数,故可以封装为一个execute方法。

select

async def select(sql, args, size=None):'''此处为选取数据库相关数据操作'''log(sql, args)global __poolasync with __pool.get() as conn:#从连接池获取一个connectasync with conn.cursor(aiomysql.DictCursor) as cur:#获取游标cursorawait cur.execute(sql.replace('?', '%s'), args or ())#将输入的sql语句中的'?'替换为具体参数argsif size:rs = await cur.fetchmany(size)else:rs = await cur.fetchall()logging.info('rows returned: %s' % len(rs))return rs

execute(update, insert, delete)

async def execute(sql, args, autocommit=True):'''此处执行数据库删减、增添等修改该操作'''log(sql)async with __pool.get() as conn:if not autocommit:await conn.begin()try:async with conn.cursor(aiomysql.DictCursor) as cur:await cur.execute(sql.replace('?', '%s'), args)affected = cur.rowcountif not autocommit:await conn.commit()except BaseException as e:if not autocommit:await conn.rollback()raisereturn affected#返回修改行

自顶向下的设计方案

当没有思路时,设想如果有一个现成的ORM框架,该去如何使用呢?

class Model(object):async def save(self, **kw):passclass User(Model):__table__ = 'users' #设定操作数据库表name = StringField(...) #设定列属性score = FloatField(...) #设定列属性passu = User(name='Mike', score=98.23)
u.save()

也就说当操作某数据库的一个数据库表时,只需创立一个类,并利用继承的方式,从父类中继承相关属性与方法,这样就可以直接完成对数据库的操作。

字段类

class Field(object):'''用于标识model每个成员变量的类name:表名称, column_type:值类型, primary_key:是否主键'''def __init__(self, name, column_type, primary_key, default):self.name = nameself.column_type = column_typeself.primary_key = primary_keyself.default = defaultdef __str__(self):return '<%s, %s:%s>' % (self.__class__.__name__, self.column_type, self.name)class StringField(Field):def __init__(self, name=None, primary_key=False, default=None, ddl='varchar(100)'):super().__init__(name, ddl, primary_key, default)class IntegerField(Field):pass...

Model类

class ModelMetaclass(type):def __new__(cls, name, bases, attrs):if name == 'Model':return type.__new__(cls, name, bases, attrs)tableName = attrs.get('__table__', None) or namelogging.info('found model: %s (table: %s)' %(name, tableName))mappings = dict()fields = []#可以理解为列名称primaryKey = Nonefor k, v in attrs.items():if isinstance(v, Field):logging.info('  found mapping: %s ==> %s' %(k, v))mappings[k] = vif v.primary_key:#判断主键并记录if primaryKey:raise RuntimeError('Duplicate primary key for field: %s' %k)primaryKey = k#记录主键else:fields.append(k)if not primaryKey:raise RuntimeError('Primary key not found.')for k in mappings.keys():attrs.pop(k)#删除attrs里属性,防止与实例属性冲突escaped_fields = list(map(lambda f: ' %s ' %f, fields))attrs['__mappings__'] = mappings  # 保存属性和列的映射关系attrs['__table__'] = tableNameattrs['__primary_key__'] = primaryKey  # 主键属性名attrs['__fields__'] = fields  # 除主键外的属性名attrs['__select__'] = 'select `%s`, %s from `%s`' % (primaryKey, ', '.join(escaped_fields), tableName)attrs['__insert__'] = 'insert into `%s` (%s, `%s`) values (%s)' % (tableName, ', '.join(escaped_fields), primaryKey, create_args_string(len(escaped_fields) + 1))attrs['__update__'] = 'update `%s` set %s where `%s`=?' % (tableName, ', '.join(map(lambda f: '`%s`=?' % (mappings.get(f).name or f), fields)), primaryKey)attrs['__delete__'] = 'delete from `%s` where `%s`=?' % (tableName, primaryKey)return type.__new__(cls, name, bases, attrs)#------------------------------------------------------------------
class Model(dict, metaclass=ModelMetaclass):def __init__(self, **kw):super(Model, self).__init__(**kw)def __getattr__(self, key):try:return self[key]except KeyError:raise AttributeError(r"'Model' object has no attribute '%s'" % key)def __setattr__(self, key, value):self[key] = valuedef getValue(self, key):return getattr(self, key, None)def getValueOrDefault(self, key):value = getattr(self, key, None)if value is None:field = self.__mappings__[key]if field.default is not None:value = field.default() if callable(field.default) else field.default#callable(obj)判断对象是否可调用logging.debug('using default value for %s: %s' %(key, str(value)))setattr(self, key, value)return value@classmethodasync def findAll(cls, where=None, args=None, **kw):'find objects by where clause'sql = [cls.__select__]if where:sql.append('where')sql.append(where)if args is None:args = []orderBy = kw.get('orderBy', None)if orderBy:sql.append('order by')sql.append(orderBy)limit = kw.get('limit', None)if limit is not None:sql.append('limit')if isinstance(limit, int):sql.append('?')args.append(limit)elif isinstance(limit, tuple) and len(limit) == 2:sql.append('?, ?')args.extend(limit)else:raise ValueError('Invalid limit value: %s' % str(limit))rs = await select(' '.join(sql), args)return [cls(**r) for r in rs]@classmethodasync def findNumber(cls, selectField, where=None, args=None):'find number by select and where'sql = ['select %s _num_ from `%s`' %(selectField, cls.__table__)]if where:sql.append('where')sql.append(where)rs = await select(' '.join(sql), args, 1)if len(rs) == 0:return Nonereturn rs[0]['_num_']@classmethodasync def find(cls, pk):'find object by primary key'rs = await select('%s where `%s`=?' %(cls.__select__, cls.__primary_key__), [pk], 1)if len(rs) == 0:return Nonereturn cls(**rs[0])async def save(self):args = list(map(self.getValueOrDefault, self.__fields__))args.append(self.getValueOrDefault(self.__primary_key__))rows = await execute(self.__insert__, args)if rows != 1:logging.warn('failed to insert record: affected rows: %s' % rows)async def update(self):args = list(map(self.getValue, self.__fields__))args.append(self.getValue(self.__primary_key__))rows = await execute(self.__update__, args)if rows != 1:logging.warn('failed to update by primary key: affected rows: %s' %rows)async def remove(self):args = [self.getValue(self.__primary_key__)]rows = await execute(self.__delete__, args)if rows != 1:logging.warn('failed to remove by primary key: affected rows: %s' % rows)
关于元类

暂时没有理解透彻,之后再做补充。

这篇关于day3——利用python创立ORM的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/476847

相关文章

Python办公自动化实战之打造智能邮件发送工具

《Python办公自动化实战之打造智能邮件发送工具》在数字化办公场景中,邮件自动化是提升工作效率的关键技能,本文将演示如何使用Python的smtplib和email库构建一个支持图文混排,多附件,多... 目录前言一、基础配置:搭建邮件发送框架1.1 邮箱服务准备1.2 核心库导入1.3 基础发送函数二、

Python包管理工具pip的升级指南

《Python包管理工具pip的升级指南》本文全面探讨Python包管理工具pip的升级策略,从基础升级方法到高级技巧,涵盖不同操作系统环境下的最佳实践,我们将深入分析pip的工作原理,介绍多种升级方... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

基于Python实现一个图片拆分工具

《基于Python实现一个图片拆分工具》这篇文章主要为大家详细介绍了如何基于Python实现一个图片拆分工具,可以根据需要的行数和列数进行拆分,感兴趣的小伙伴可以跟随小编一起学习一下... 简单介绍先自己选择输入的图片,默认是输出到项目文件夹中,可以自己选择其他的文件夹,选择需要拆分的行数和列数,可以通过

Python中反转字符串的常见方法小结

《Python中反转字符串的常见方法小结》在Python中,字符串对象没有内置的反转方法,然而,在实际开发中,我们经常会遇到需要反转字符串的场景,比如处理回文字符串、文本加密等,因此,掌握如何在Pyt... 目录python中反转字符串的方法技术背景实现步骤1. 使用切片2. 使用 reversed() 函

Python中将嵌套列表扁平化的多种实现方法

《Python中将嵌套列表扁平化的多种实现方法》在Python编程中,我们常常会遇到需要将嵌套列表(即列表中包含列表)转换为一个一维的扁平列表的需求,本文将给大家介绍了多种实现这一目标的方法,需要的朋... 目录python中将嵌套列表扁平化的方法技术背景实现步骤1. 使用嵌套列表推导式2. 使用itert

使用Docker构建Python Flask程序的详细教程

《使用Docker构建PythonFlask程序的详细教程》在当今的软件开发领域,容器化技术正变得越来越流行,而Docker无疑是其中的佼佼者,本文我们就来聊聊如何使用Docker构建一个简单的Py... 目录引言一、准备工作二、创建 Flask 应用程序三、创建 dockerfile四、构建 Docker

Python使用vllm处理多模态数据的预处理技巧

《Python使用vllm处理多模态数据的预处理技巧》本文深入探讨了在Python环境下使用vLLM处理多模态数据的预处理技巧,我们将从基础概念出发,详细讲解文本、图像、音频等多模态数据的预处理方法,... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

Python使用pip工具实现包自动更新的多种方法

《Python使用pip工具实现包自动更新的多种方法》本文深入探讨了使用Python的pip工具实现包自动更新的各种方法和技术,我们将从基础概念开始,逐步介绍手动更新方法、自动化脚本编写、结合CI/C... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

Conda与Python venv虚拟环境的区别与使用方法详解

《Conda与Pythonvenv虚拟环境的区别与使用方法详解》随着Python社区的成长,虚拟环境的概念和技术也在不断发展,:本文主要介绍Conda与Pythonvenv虚拟环境的区别与使用... 目录前言一、Conda 与 python venv 的核心区别1. Conda 的特点2. Python v

Python使用python-can实现合并BLF文件

《Python使用python-can实现合并BLF文件》python-can库是Python生态中专注于CAN总线通信与数据处理的强大工具,本文将使用python-can为BLF文件合并提供高效灵活... 目录一、python-can 库:CAN 数据处理的利器二、BLF 文件合并核心代码解析1. 基础合