摘自以前CI实现的商城系统，做APP时需要实现数据接口，便用python实现了。
假设有表tp_article
| id | title | type |
| 1 | 哈哈 | 1 |
| 2 | 图样涂森坡 | 1 |
使用thinphp实现取出type为1的数据如下
M()->from(‘‘tp_article‘‘)->where(‘type=1‘)->select();
现通过python实现类似对mysql进行操作的封装
DB.select(‘*‘).fm(‘tp_article‘).where(‘type‘,1).all()
需要：django下的db模块
首先实现 mydb.py 文件，放于core目录下，实现模型类文件时需要import
from django.db import connection
cursor = connection.cursor()
'''
Simple function for one result sql
'''
#返回一条结果
def fetchone(sql):
cursor.execute(sql)
try:
col_names = [row[0] for row in cursor.description]
rs = dict(zip(col_names,cursor.fetchone()))
except:
rs = {}
return rs
<pre name="code" class="python">#返回多条结果
def fetchall(sql): cursor.execute(sql) try: col_names = [row[0] for row in cursor.description] data = cursor.fetchall() rs = [dict(zip(col_names,raw)) for raw in data] except: rs=[] return rs
然后就是实现模型类文件db_mysql.py
from core.mydb import *
class Db_mysql(object):
'''build sql'''
sql=''
options={}
history=[]
def select(self,fields):
select_list = fields.split(',')
self.options['select']=select_list
return self
#实现from，因为from是python保留关键词，所以用fm代替
def fm(self,table):
self.options['from'] = table
return self
def where(self,seg,*args):
'''
sql query of where tag
'''
try:
where_list=self.options['where']
except:
where_list=[]
if isinstance(seg,str):
try:
where_list.append({'key':seg,'val':args[0],'type':args[1]})
except:
where_list.append({'key':seg,'val':args[0],'type':False})
elif isinstance(seg,dict):
for key in seg.keys():
where_list.append({'key':key,'val':seg[key]})
self.options['where']= where_list
return self
def where_in(self,key,val):
if not isinstance(self.options['where'],list):
self.options['where']=[]
self.options['where'].append({'key':key,'val':str(val),'type':'in'})
return self
def like(self,key,words):
if not isinstance(self.options['where'],list):
self.options['where']=[]
self.options['where'].append({'key':key,'val':str(words),'type':'like'})
return self
def join(self,table,on,type='left'):
self.options['join']= {'table':table,'on':on,'type':type}
return self
def limit(self,offset=0,size=0):
offset = int(offset)
size = int(size)
if size == 0:
size=offset
offset=0
self.options['limit']= {'offset':str(offset),'size':str(size)}
return self
def order(self,oby,desc='desc'):
self.options['order']={'order':str(oby),'desc':str(desc)}
return self
#组装sql就靠这个方法了
def combile_sql(self):
'''
combile select sql
'''
if not isinstance(self.options['select'],list):
self.options['select']=['*']
self.sql=''.join(['select '])
length = len(self.options['select'])
for i in range(length):
fields = self.options['select'][i]
if i==length-1:
self.sql=''.join([self.sql,fields,' '])
break
self.sql=''.join([self.sql,fields,', '])
'''
combile from sql
'''
self.sql=''.join([self.sql,' from ',self.options['from'],' '])
'''
combile join sql
'''
try:
if isinstance(self.options['join'],dict):
join_str = self.options['join']
except:
pass
else:
self.sql=''.join([self.sql,join_str['type'],' join ',join_str['table'],' on ',join_str['on'],' '])
'''
combile where sql and where in sql
'''
try:
where_list = self.options['where']
except:
where_list = []
else:
if len(where_list):
self.sql=''.join([self.sql,' where '])
count=0
for item in where_list:
if count is 0:
segment = ' '
else:
segment = ' AND '
count=count+1
if not item.get('type',False):
self.sql=''.join([self.sql,segment,item['key'],'=',str(item['val'])])
elif item['type'] is 'in':
self.sql=''.join([self.sql,segment,item['key'],' in (',str(item['val']),') '])
elif item['type'] is 'like':
self.sql=''.join([self.sql,segment,item['key'],' like %',str(item['val']),'% '])
'''
combile order sql
'''
try:
order_dict = self.options['order']
if isinstance(order_dict,dict):
self.sql=''.join([self.sql,' order by ',order_dict['order'],' ',order_dict['desc']])
except:
pass
'''
combile limit sql
'''
try:
self.sql=''.join([self.sql,' limit ',self.options['limit']['offset'],',',self.options['limit']['size']])
self.history.append(self.sql)
self.options = {}
except KeyError:
pass
return self
#查询操作，类似tp李的find
def get(self,table=False):
if not isinstance(table,bool):
self.options['from']
self.combile_sql()
rs={}
try:
rs = fetchone(self.sql)
except Exception as e:
print e
print self.sql
self.sql=''
return rs
#查询操作，类似tp里的select
def all(self,table=False):
if not isinstance(table,bool):
self.options['from']
self.combile_sql()
rs = []
try:
rs = fetchall(self.sql)
except Exception as e:
print e
print self.sql
self.sql=''
return rs
#修改操作
def update(self,table=False,*args):
if not isinstance(table,bool):
self.sql=''.join(['update ',table,' set '])
else:
return False
<span style="white-space:pre"> </span>#判断方法接收的参数是字符串还是字典，做不同处理
if isinstance(args[0],str):
if isinstance(args[1],str):
val = ''.join(["'",args[1],"'"])
else:
val = str(args[1])
self.sql = ''.join([self.sql,args[0],'=',val,' '])
elif isinstance(args[0],dict):
count=0
for key in args[0].keys():
if count is 0:
segment = ' '
else:
segment = ','
if isinstance(args[0][key],str):
val = ''.join(["'",args[0][key],"'"])
else:
val = str(args[0][key])
self.sql = ''.join([self.sql,segment,key,'=',val,' '])
count = count+1
'''
combile where sql and where in sql
'''
try:
where_list = self.options['where']
except:
where_list = []
else:
if len(where_list):
self.sql=''.join([self.sql,' where '])
count=0
for item in where_list:
if count is 0:
segment = ' '
else:
segment = ' AND '
count=count+1
if not item.get('type',False):
self.sql=''.join([self.sql,segment,item['key'],'=',str(item['val'])])
elif item['type'] is 'in':
self.sql=''.join([self.sql,segment,item['key'],' in (',str(item['val']),') '])
elif item['type'] is 'like':
self.sql=''.join([self.sql,segment,item['key'],' like %',str(item['val']),'% '])
'''
combile where sql and where in sql
'''
rs = fetchone(self.sql)
self.options = {}
self.sql=''
return rs
#插入操作
def insert(self,table=False,seg={}):
if not isinstance(table,bool):
self.sql=''.join(['insert into ',table,' '])
else:
return False
if isinstance(seg,dict):
count=0
keystr=''
valstr=''
for key in seg.keys():
if count is 0:
segment = ''
else:
segment = ','
keystr = ''.join([keystr,segment,key])
if isinstance(seg[key],str):
val = ''.join(["'",seg[key],"'"])
else:
val = str(seg[key])
valstr = ''.join([valstr,segment,val])
self.sql=''.join([self.sql,'(',keystr,')',' values ','(',valstr,')'])
rs = fetchone(self.sql)
self.options = {}
self.sql=''
return rs
#删除操作
def delete(self,table=False):
if not isinstance(table,bool):
self.sql=''.join(['delete from ',table,' '])
else:
return False
'''
combile where sql and where in sql
'''
try:
where_list = self.options['where']
except:
where_list = []
else:
if len(where_list):
self.sql=''.join([self.sql,' where '])
count=0
for item in where_list:
if count is 0:
segment = ' '
else:
segment = ' AND '
count=count+1
if not item.get('type',False):
self.sql=''.join([self.sql,segment,item['key'],'=',str(item['val'])])
elif item['type'] is 'in':
self.sql=''.join([self.sql,segment,item['key'],' in (',str(item['val']),') '])
elif item['type'] is 'like':
self.sql=''.join([self.sql,segment,item['key'],' like %',str(item['val']),'% '])
'''
combile where sql and where in sql
'''
rs = fetchone(self.sql)
self.options = {}
self.sql=''
return rs
#测试
DB = Db_mysql()
DB.select('attr_id,attr_name').fm('hr_attribute').where('attr_type',0).where_in('attr_id','3,4,6').join('goods as g','g.id=a.id','left').order('attr_id').limit(3,5);
DB.get()
经测试，没有什么大问题，如果有问题建议请告诉我哦
时间： 2024-10-15 10:57:50