【国产RISC-V Linux板 昉·星光VisionFive试用报告】创建用户数据插入
<p>tornado创建模型用于数据库的插入，今天设计被采样人数据插入，设计models下面的model.py</p><pre>
<code>from peewee import *
import peewee_async
database = peewee_async.MySQLDatabase(
host='localhost',
port=3306,
user='user',
password='pypasswd',
database='test_db',
autocommit=False
)
objects = peewee_async.Manager(database)
database.set_allow_sync(True)
class BaseModel(Model):
class Meta:
database = database
class personnel_information(BaseModel):
name = CharField(max_length=10, verbose_name="姓名", index=True)
ID_card = CharField(max_length=20, verbose_name="身份证号")
phone = CharField(max_length=15, verbose_name="手机号码")
uint = CharField(max_length=45, verbose_name="单位名称")
class specimen_information(BaseModel):
tube_Id = CharField(max_length=10, verbose_name="试管号", index=True)
name = CharField(max_length=10, verbose_name="姓名")
phone = CharField(max_length=15, verbose_name="手机号")
uint = CharField(max_length=45, verbose_name="单位名称")
acquisition_time = DateTimeField(verbose_name="采样时间")
Collector = CharField(max_length=10, verbose_name="采样人姓名")
Collection_unit = CharField(max_length=40, verbose_name="采样单位")
</code></pre>
<p>wiews_index.py下面新建class SamplePersnnelHandler</p>
<pre>
<code>class SamplePersnnelHandler(CommonHandler):
def check_xsrf_cookie(self):
return True
async def post(self):
res = dict(code=9999, msg='')
print(self.request.body)
try:
""" 分析POST请求参数 接收消息 获取参数"""
data_quest = json.loads(self.request.body)
if 'command' not in data_quest:
res['msg'] = "请救命令参数错误"
else:
print(data_quest['command'])
print(data_quest['name'])
if data_quest['command'] == "appendPersonnel":
# 插入被采人信息
if 'name' not in data_quest:
res['msg'] = '姓名不能为空'
if 'ID_card' not in data_quest:
res['msg'] = res['msg'] + '身份证号码为空'
if 'phone' not in data_quest:
res['msg'] = res['msg'] + "手机号码为空"
if 'uint' not in data_quest:
res['msg'] = res['msg'] + "单位为空"
if res['msg'] == "":
# 开始插入采样人信息
print("开始保存数据")
query = personnel_information.select().where(personnel_information.name == data_quest['name'])
results = await objects.execute(query.dicts())
if results:
print("用户已存在")
res['code'] = 2001
res['msg'] = "用户已存在"
else:
try:
ob = await objects.create(
personnel_information,
name=data_quest['name'],
ID_card=data_quest['ID_card'],
phone=data_quest['phone'],
uint=data_quest['uint']
)
if ob:
res['code'] = 1000
res['msg'] = "save OK"
else:
res['code'] = 2001
res['msg'] = "save data err"
except Exception as e:
res['code'] = 2003
res['msg'] = "save data ERROR"
else:
res['msg'] = "command err!"
except Exception as e:
print("错误信息：" + str(e))
print(res)
self.write(res)</code></pre>
<p>增加路由：</p>
<pre>
<code>from app.api.views_index import IndexHandler as api_index
from app.api.views_index import SamplePersnnelHandler as man_index
# API接口
api_urls = [
(r'/', api_index),
(r'/man', man_index),
]</code></pre>
<p>重启服务后用curl模拟POST插入数据：</p>
<p> 然后数据库更新了数据：</p>
<p> 数据插入成功。下面书写数据读取、采样信息的插入。这段时间主要设计数据库的东西多一些，要等服务器建好才能进入昉星光的程序展示，大家耐心等待。</p>
页:
[1]