flashkはsqlalchemyを使ってsqliteデータベースを添削して調べます.
7849 ワード
flashkはsqlalchemyを使ってsqliteデータベースを添削してコードを調べます.
https://www.cnblogs.com/eating-gourd/p/9997751.html
#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
@author:ZSW
@file:test4.py
@time:2020/04/27
"""
from flask import Flask
import json
from sqlite3test.conf.GetConfParams import GetConfParams
import pandas as pd
from sqlite3test.tojson import queryToDict
ConfParams = GetConfParams()
logger = ConfParams.logger #
app = Flask(__name__)
from flask_sqlalchemy import SQLAlchemy
### sqlalchemy
#
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///E:/sqlite/sqlite-tools-win32-x86-3310100/study.db'
# True ( )，Flask-SQLAlchemy 。 ， 。
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
###
db = SQLAlchemy(app)
class Student(db.Model):
__tablename__ = "student"
id = db.Column(db.Integer,primary_key=True)
name = db.Column(db.String(32))
age = db.Column(db.String(32))
gender = db.Column(db.String(32))
classes = db.Column(db.String(32))
# def __repr__(self): # __repr__ ，
# return {'name':'{}'.format(self.name)}
db.create_all() #
# db.drop_all() #
###
#
def insertData():
try:
# student = Student(name='lee',age=18,gender='man',classes='python') #
# tudent)
db.session.commit()
except Exception as e:
logger.error(e)
res = {'code': 0, 'message': ' '}
return json.dumps(res, ensure_ascii=False, indent=4)
#
def patchinsertData():
try:
student1 = Student(name=' ',age=18,gender='male',classes='java')
student2 = Student(name=' ',age=19,gender='female',classes='c++')
student3 = Student(name=' ',age=22,gender='male',classes='php')
student4 = Student(name=' ',age=25,gender='female',classes='matalab')
db.session.add_all([student1,student2,student3,student4])
db.session.commit()
except Exception as e:
logger.error(e)
res = {'code': 0, 'message': ' '}
return json.dumps(res, ensure_ascii=False, indent=4)
#
def deleteData():
try:
student = Student.query.get(2) # id 2
db.session.delete(student)
db.session.commit()
except Exception as e:
logger.error(e)
res = {'code': 0, 'message': ' '}
return json.dumps(res, ensure_ascii=False, indent=4)
#
def updateData():
try:
student = Student.query.get(7) # id 2
student.name = ' '
student.age = 24
student.gender = 'male'
db.session.commit()
except Exception as e:
logger.error(e)
res = {'code': 0, 'message': ' '}
return json.dumps(res, ensure_ascii=False, indent=4)
#
def selectData():
try:
student = Student.query.all()
print(student) #print(student) name ， __repr__(self):return self.name
for s in student:
print(s.name)
except Exception as e:
logger.error(e)
res = {'code': 0, 'message': ' '}
return json.dumps(res, ensure_ascii=False, indent=4)
#
def deleteallData():
try:
student = Student.query.all()
print(student) #print(student) name ， __repr__(self):return self.name
for s in student:
db.session.delete(s) #
db.session.commit()
except Exception as e:
logger.error(e)
res = {'code': 0, 'message': ' '}
return json.dumps(res, ensure_ascii=False, indent=4)
# ，
def checkData():
try:
# ：
student = (db.session.query(Student.name,Student.id,Student.age).filter(Student.age==18).all())
print(student)
res = [dict(zip(r.keys(), r)) for r in student] #
print(res)
# ：
# student = Student.query.filter_by(age=18)
# # print(student) # student SQL
# for s in student:
# # print(s)
# result = queryToDict(s)
# print(result)
# # print(s.id, s.name, s.classes)
except Exception as e:
logger.error(e)
res = {'code': 0, 'message': ' '}
return json.dumps(res, ensure_ascii=False, indent=4)
# ：
"""
student = Student.query.filter_by(age = 18) #
student = Student.query.get(1) #
student = Student.query.group_by("age") #
#
student = Student.query.order_by(Student.age)
#
student = Student.query.order_by(Student.age.desc()) #
"""
if __name__ == '__main__':
checkData()
import logging
import logging.config
import os
class GetConfParams:
PATH = os.getcwd() #
logging.config.fileConfig(PATH + '/conf/logging.conf')
def __init__(self):
self.logger = logging.getLogger('root')
#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
@author:ZSW
@file:tojson.py
@time:2020/04/27
"""
#flask_sqlalchemy dict
from datetime import datetime as cdatetime # datatime
from datetime import date, time
from flask_sqlalchemy import Model
from sqlalchemy.orm.query import Query
from sqlalchemy import DateTime, Numeric, Date, Time # DateTime
def queryToDict(models):
if (isinstance(models, list)):
if (isinstance(models[0], Model)):
lst = []
for model in models:
gen = model_to_dict(model)
dit = dict((g[0], g[1]) for g in gen)
lst.append(dit)
return lst
else:
res = result_to_dict(models)
return res
else:
if (isinstance(models, Model)): #####************** 1 *************************
gen = model_to_dict(models)
dit = dict((g[0], g[1]) for g in gen)
return dit ##### 4
else:
res = dict(zip(models.keys(), models))
find_datetime(res)
return res
# result ，result key()
def result_to_dict(results):
res = [dict(zip(r.keys(), r)) for r in results]
# r ，
for r in res:
find_datetime(r)
return res
def model_to_dict(model): # ##### 2
for col in model.__table__.columns:
if isinstance(col.type, DateTime):
value = convert_datetime(getattr(model, col.name))
elif isinstance(col.type, Numeric):
value = float(getattr(model, col.name))
else:
value = getattr(model, col.name) ##### 3
yield (col.name, value)
def find_datetime(value):
for v in value:
if (isinstance(value[v], cdatetime)):
value[v] = convert_datetime(value[v]) # ， ，
def convert_datetime(value):
if value:
if (isinstance(value, (cdatetime, DateTime))):
return value.strftime("%Y-%m-%d %H:%M:%S")
elif (isinstance(value, (date, Date))):
return value.strftime("%Y-%m-%d")
elif (isinstance(value, (Time, time))):
return value.strftime("%H:%M:%S")
else:
return ""
https://www.cnblogs.com/eating-gourd/p/9997751.html