ORM框架实现 & sqlalchemy
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了ORM框架实现 & sqlalchemy，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含9912字，纯文字阅读大概需要15分钟。
内容图文
import pymysql
from pymysql.cursors import DictCursor
# class Field:
# def __init__(self,name,column=None,chief=False,unique=False,index=False,nullable=True,default=None):
# self.name=name
# if column is None:
# self.column=name
# else:
# self.column=column
# self.chief=chief
# self.unique=unique
# self.index=index
# self.nullable=nullable
# self.default=default
#
# def validate(self,value):
# raise NotImplementedError
#
# def __get__(self,instance,owner):
# if instance is None:
# return self
# return instance.__dict__[self.name]
#
# def __set__(self,instance,value):
# self.validate(value)
# instance.__dict__[self.name]=value
#
# def __str__(self):
# return '<{} <{}>>'.format(self.__class__.__name__,self.name)
#
# __repr__=__str__
#
# class IntField(Field):
# def __init__(self,name,column=None,chief=False,unique=False,index=False,nullable=True,default=None,
# auto_increment=False):
# super(IntField,self).__init__(name,column,chief,unique,index,nullable,default)
# self.auto_increment=auto_increment
#
# def validate(self,value):
# if value is None:
# if self.chief:
# raise TypeError('{} is primary key,yet value is None'.format(self.name))
# if not self.nullable:
# raise TypeError('{} required'.format(self.name))
# else:
# if not isinstance(value,int):
# raise GeneratorExit('{} should be integer'.format(value))
#
# class StringField(Field):
# def __init__(self,name,column=None,chief=None,unique=None,index=None,nullable=True,default=None,length=None):
# super(StringField,self).__init__(name,column,chief,unique,index,nullable,default)
# self.length=length
#
# def validate(self,value):
# if value is None:
# if self.chief:
# raise TypeError('{} is primary key,yet value is None'.format(self.name))
# if not self.nullable:
# raise TypeError('{} required'.format(self.name))
# else:
# if not isinstance(value, str):
# raise GeneratorExit('{} should be str'.format(value))
# if len(value) > self.length:
# raise ValueError('{} too long'.format(value))
#
# class S:
# id=IntField('id','id',True,nullable=False,auto_increment=True)
# name=StringField('name',nullable=False,length=64)
# age=IntField('age')
#
# def __init__(self,id,name,age):
# self.id=id
# self.name=name
# self.age=age
#
# def __str__(self):
# return 'S({} {} {})'.format(self.id,self.name,self.age)
#
# __repr__=__str__
#
# def save(self,conn:pymysql.connections.Connection):
# sql='insert into s (id,name,age) values (%s,%s,%s)'
# cursor=conn.cursor(cursor=DictCursor)
# cursor.execute(sql,args=(self.id,self.name,self.age))
# cursor.close()
class Field:
def __init__(self, name=None, column=None, chief=False, unique=False, index=False, nullable=True, default=None):
self.name = name
if column is None:
self.column = name
else:
self.column = column
self.chief = chief
self.unique = unique
self.index = index
self.nullable = nullable
self.default = default
def validate(self, value):
raise NotImplementedError()
def __get__(self, instance, owner):
if instance is None:
return self
return instance.__dict__[self.name]
def __set__(self, instance, value):
self.validate(value)
instance.__dict__[self.name] = value
def __str__(self):
return '<{} {}>'.format(self.__class__.__name__, self.name)
__repr__ = __str__
class IntField(Field, object):
def __init__(self, name=None, column=None, chief=False, unique=False, index=False, nullable=True, default=None,
auto_increment=False):
self.auto_increment = auto_increment
super(IntField, self).__init__(name, column, chief, unique, index, nullable, default)
def validate(self, value):
if value is None:
if self.chief:
raise TypeError('{} is primary key,yet value is None'.format(self.name))
if not self.nullable:
raise TypeError('{} required'.format(self.name))
else:
if not isinstance(value, int):
raise GeneratorExit('{} should be integer'.format(value))
class StringField(Field):
def __init__(self, length, name=None, column=None, chief=False, unique=False, index=False, nullable=True,
default=None, ):
self.length = length
super(StringField, self).__init__(name, column, chief, unique, index, nullable, default)
def validate(self, value):
if value is None:
if self.chief:
raise TypeError('{} is primary key,yet value is None'.format(self.name))
if not self.nullable:
raise TypeError('{} required'.format(self.name))
else:
if not isinstance(value, str):
raise GeneratorExit('{} should be str'.format(value))
if len(value) > self.length:
raise ValueError('{} too long'.format(value))
class Session:
def __init__(self, conn: pymysql.connections.Connection):
self.conn = conn
self.cursor = None
self.mark = False
def execute(self, sql, *args, **kwargs):
if self.cursor is None:
self.mark = True
self.cursor = self.conn.cursor(cursor=DictCursor)
if args:
self.cursor.execute(sql, args)
if kwargs:
self.cursor.execute(sql, args=kwargs)
if self.mark:
self.conn.commit()
self.mark = False
def __enter__(self):
self.cursor = self.conn.cursor(cursor=DictCursor)
return self # 以后调用session对象的execute
def __exit__(self, exc_type, exc_value, exc_tb):
self.cursor.close()
print(exc_type, exc_value, exc_tb)
if exc_type:
self.conn.rollback()
else:
self.conn.commit()
# class Session:
# def __init__(self, conn: pymysql.connections.Connection):
# self.conn = conn
#
# def execute(self, sql, *args, **kwargs):
# try:
# cursor = self.conn.cursor(cursor=DictCursor)
# with cursor:
# if args:
# cursor.execute(sql, args)
# if kwargs:
# cursor.execute(sql, kwargs)
# except:
# self.conn.rollback()
# else:
# self.conn.commit()
class ModelMeta(type):
def __new__(cls, what: str, bases, attrs: dict):
# print(what,bases,attrs)
if attrs.get('__tablename__', None) is None:
# if '__tablename__' not in attrs.keys():
attrs.setdefault('__tablename__', what.lower())
mapping = {}
primary_key = []
for k, v in attrs.items():
if isinstance(v, (Field,)):
if v.name is None:
v.name = k # descriptor
if v.column is None:
v.column = k
mapping[k] = v
if v.chief:
primary_key.append(v)
attrs.setdefault('__mapping__', mapping)
attrs.setdefault('__primary_key__', primary_key)
return super(ModelMeta, cls).__new__(cls, what, bases, attrs) # 需要return,否则Model为None
class Model(metaclass=ModelMeta):
def save(self, session: Session):
names = []
columns = []
values = []
for k, v in type(self).__dict__.items(): # 遍历self实例字典,v是值,而type(self)字典为描述器,借助描述器得到column
if isinstance(v, (Field,)):
if k in self.__dict__.keys():
names.append(k)
columns.append(v.column)
values.append(self.__dict__[k])
print(names, columns, values)
sql = 'insert into {} ({}) values ({})'.format(
self.__tablename__,
','.join(columns),
','.join(['%s'] * len(columns))
)
with session:
session.execute(sql, *values)
class S(Model):
__tablename__ = 'pp'
id = IntField('id', column='id', chief=True, nullable=False, auto_increment=True)
sname = StringField(64, 'sname', column='name', nullable=False)
age = IntField('age', chief=True)
def __init__(self, id, name, age):
self.id = id
self.sname = name
self.age = age
def __str__(self):
return 'Student({},{},{})'.format(self.id, self.sname, self.age)
__repr__ = __str__
# def save(self, conn: pymysql.connections.Connection):
# sql = 'insert into pp (id,name,age) values (%(id)s,%(name)s,%(age)s)'
# try:
# cursor = conn.cursor(cursor=DictCursor)
# with cursor:
# cursor.execute(sql, args={'id': self.id, 'name': self.name, 'age': self.age})
# conn.commit()
# except:
# conn.rollback()
#
# def save(self, session: Session):
# sql = 'insert into pp (id,name,age) values (%(id)s,%(name)s,%(age)s)'
# session.execute(sql, id=self.id, name=self.name, age=self.age)
# # with session:
# # session.execute(sql,id=self.id,name=self.name,age=self.age)
#
# def save(self, session: Session):
# sql = 'insert into pp (id,age,name) values (%(id)s,%(age)s,%(name)s)'
# session.execute(sql, name=self.name, age=self.age, id=self.id)
class Engine:
def __init__(self,*args,**kwargs):
self.conn=pymysql.connect(*args,**kwargs)
def save(self,instance:S):
names=[]
columns=[]
values=[]
for k,v in instance.__mapping__.items():
if k in instance.__dict__.keys():
names.append('`{}`'.format(k))
columns.append('`{}`'.format(v.column))
values.append(instance.__dict__[k])
query='insert into {} ({}) values ({})'.format(
instance.__tablename__,
','.join(columns),
','.join(['%s']*len(columns))
)
try:
cursor=self.conn.cursor(cursor=DictCursor)
with cursor:
cursor.execute(query,args=values)
except Exception as e:
print(e.args,e.__context__)
self.conn.rollback()
else:
self.conn.commit()
conn = pymysql.connections.Connection('localhost', 'root', 'cruces', 'uranus')
s = S(444, 'zxc', 88)
print(s.id, s.sname, s.age)
# s.save(Session(conn))
print(s.__dict__)
print(S.__dict__)
engine=Engine('localhost', 'root', 'cruces', 'uranus')
engine.save(s)
# for k,v in S.__dict__.items():
# # print(v,v.__class__.__bases__,type(v.__class__))
# if v.__class__.__bases__[0] == Field:
# print(k)
import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Float,Enum,Date
Base=declarative_base()
class Student(Base):
__tablename__='Stu'
id=Column(Integer,primary_key=True,nullable=False,autoincrement=True)
name=Column(String(64),nullable=False)
age=Column(Float,unique=True)
def __repr__(self):
return '<{} id:{}, name:{}, age:{}>'.format(self.__class__.__name__,self.id,self.name,self.age)
__str__=__repr__
engine=sqlalchemy.create_engine("mysql+pymysql://root:cruces@localhost:3306/uranus",echo=True)
Base.metadata.create_all(engine)
内容总结
以上是互联网集市为您收集整理的ORM框架实现 & sqlalchemy全部内容，希望文章能够帮你解决ORM框架实现 & sqlalchemy所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】