备案控制台
探索云世界
开发者社区 开发与运维 文章 正文

python tornado 实现类禅道系统

2022-04-21 132
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 笔记

背景：目前市面上有很多bug管理工具，但是各有各的特点，最著名，最流行的就是禅道，一个偶然的机会接触到了python ，学到tornado后，就想着去怎么去用到实处，后来发现自己公司的除了禅道就记录bug没有什么可以用的工具了。

语言：python3 第三库 ：tornado，qiniu(用于云存储文件)，数据库用sqlite

why  use tornado？很多人其实会这么问我，我感觉tornado可以实现异步，虽然现在代码还没有用到异步，我感觉还是很不错的框架，值得学习，现在很多公司都在用，个人感觉这是一个不错的，值得我们大家去学习的框架。

来看看我的需求文档

13.png

大题是这么的轮廓，那么拿到这个的时候，我会进行需求分析，

需要什么样的数据库， 数据模型之间的关系，虽然现在还是有很多地方是写死的还没有进行搜索功能的设置，但是我相信，有了现在这个demo

那么我开始来设计下我主要会结构

14.png

handlsers 存放类似flask的views

models存数据库相关的，

static存放静态

template存放模板

untils存放公共库

setting 配置文件

urls url地址映射

run 运行文件。

那么我来开始设计我的数据，其实我的数据模型也是一波三折的。

选择用了sqlalchemy，之前用过flask的sqlalchemy感觉不错

from models.dataconfig import db_session,Base,create_all
from sqlalchemy import  Column,Integer,DateTime,Boolean,String,ForeignKey,desc,asc,Text
from sqlalchemy.orm import  relationship,backref
from untils.common import encrypt
import datetime
class User(Base):
    __tablename__='users'
    id=Column(Integer(),primary_key=True)
    username=Column(String(64),unique=True,index=True)
    email=Column(String(64))
    password=Column(String(64))
    last_logtime=Column(DateTime())
    status=Column(Integer())
    leves=Column(Integer())
    iphone=Column(Integer())
    Projects=relationship('Project',backref='users')
    shebei=relationship('Shebei',backref='users')
    file=relationship('FilePan',backref='users')
    banben=relationship('BanbenWrite',backref='users')
    testresult=relationship('TestResult',backref='users')
    testcase=relationship('TestCase',backref='users')
    buglog=relationship('BugLog',backref='users')
    def __repr__(self):
        return self.username
    @classmethod
    def get_by_id(cls, id):
        item = db_session.query(User).filter(User.id==id).first()
        return item
    @classmethod
    def get_by_username(cls, username):
        item = db_session.query(User).filter(User.username== username).first()
        return item
    @classmethod
    def get_count(cls):
        return db_session.query(Shebei).count()
    @classmethod
    def add_new(cls,username,password,iphone,email,leves):
        new=User(username=username,iphone=iphone,email=email,leves=leves)
        new.password=encrypt(password)
        new.status=0
        db_session.add(new)
        try:
            db_session.commit()
        except:
            db_session.rollback()
class Shebei(Base):
    __tablename__='shebeis'
    id=Column(Integer(),primary_key=True)
    shebei_id=Column(String(32),unique=True)
    shebei_name=Column(String(64))
    shebei_xitong=Column(String(64))
    shebei_xinghao=Column(String(255))
    shebei_jiage=Column(Integer())
    shebei_fapiaobianhao=Column(String(64))
    shebei_quanxian=Column(Boolean())
    shebei_jie=Column(Boolean())
    shebei_shuyu=Column(String())
    shebei_date=Column(DateTime())
    shebei_user=Column(String())
    gou_date=Column(DateTime())
    shebei_status=Column(String(16))
    she_sta=Column(Integer(),default=0)
    ruku_user=Column(Integer(),ForeignKey('users.id'))
    def __repr__(self):
        return self.shebei_name
    @classmethod
    def get_by_name(cls,name):
        item=db_session.query(Shebei).filter(Shebei.shebei_name==name).first()
        return item
    @classmethod
    def get_by_id(cls,id):
        item=db_session.query(Shebei).filter(Shebei.id==id).first()
        return item
    @classmethod
    def get_count(cls):
        return db_session.query(Shebei).count()
class TestResult(Base):
    __tablename__='testresults'
    id=Column(Integer(),primary_key=True)
    porject_id=Column(Integer(),ForeignKey('projects.id'))
    creat_time=Column(DateTime())
    bug_first=Column(Integer())
    ceshirenyuan=Column(String(255))
    is_send=Column(Boolean(),default=True)
    filepath=Column(String(64))
    status=Column(Integer(),default=0)
    user_id=Column(Integer(),ForeignKey('users.id'))
    def __repr__(self):
        return self.porject_name
    @classmethod
    def get_by_name(cls,name):
        item=db_session.query(TestResult).filter(TestResult.porject_name==name).first()
        return item
    @classmethod
    def get_by_id(cls,id):
        item=db_session.query(TestResult).filter(TestResult.id==id).first()
        return item
    @classmethod
    def get_by_user_id(cls,user_id):
        item=db_session.query(TestResult).filter(TestResult.user_id==user_id).first()
        return item
    @classmethod
    def get_count(cls):
        return db_session.query(TestResult).count()
class BanbenWrite(Base):
    __tablename__='banbens'
    id=Column(Integer(),primary_key=True)
    porject_id=Column(Integer(),ForeignKey('projects.id'))
    creat_time=Column(DateTime(),default=datetime.datetime.now())
    banbenhao=Column(String(32))
    is_xian=Column(Boolean(),default=False)
    is_test=Column(Boolean(),default=False)
    status=Column(Integer())
    user_id=Column(Integer(),ForeignKey('users.id'))
    bugadmin=relationship('BugAdmin',backref='banbens')
    def __repr__(self):
        return self.banbenhao
    @classmethod
    def get_by_name(cls,name):
        item=db_session.query(BanbenWrite).filter(BanbenWrite.porject_name==name).first()
        return item
    @classmethod
    def get_by_id(cls,id):
        item=db_session.query(BanbenWrite).filter(BanbenWrite.id==id).first()
        return item
    @classmethod
    def get_by_user_id(cls,user_id):
        item=db_session.query(BanbenWrite).filter(BanbenWrite.user_id==user_id).first()
        return item
    @classmethod
    def get_count(cls):
        return db_session.query(BanbenWrite).count()
class FilePan(Base):
    __tablename__='files'
    id=Column(Integer(),primary_key=True)
    file_fenlei=Column(String(64))
    file_name=Column(String(64))
    down_count=Column(Integer(),default=0)
    creat_time=Column(DateTime(),default=datetime.datetime.now())
    status=Column(Integer(),default=0)
    down_url=Column(String(64))
    is_tui=Column(Boolean(),default=False)
    user_id=Column(Integer(),ForeignKey('users.id'))
    def __repr__(self):
        return self.file_name
    @classmethod
    def get_by_file_name(cls,name):
        item=db_session.query(FilePan).filter(FilePan.file_name==name).first()
        return item
    @classmethod
    def get_by_id(cls,id):
        item=db_session.query(FilePan).filter(FilePan.id==id).first()
        return item
    @classmethod
    def get_by_user_id(cls,user_id):
        item=db_session.query(FilePan).filter(FilePan.user_id==user_id).first()
        return item
    @classmethod
    def get_count(cls):
        return db_session.query(FilePan).count()
class BugAdmin(Base):
    __tablename__='bugadmins'
    id=Column(Integer(),primary_key=True)
    porject_id=Column(Integer(),ForeignKey('projects.id'))
    bugname=Column(String(64))
    bugdengji=Column(String(64))
    bugtime=Column(DateTime(),default=datetime.datetime.now())
    bug_miaoshu=Column(String(255))
    ban_id=Column(Integer(),ForeignKey('banbens.id'))
    fujian=Column(String(64))
    is_que=Column(Boolean())
    bug_status=Column(String(64))
    bug_jiejuefangan=Column(String(64))
    bug_send=Column(String(64))
    status=Column(Integer(),default=0)
    bug_log=relationship('BugLog',backref='bugadmins')
    user_id=Column(Integer(),ForeignKey('users.id'))
    def __repr__(self):
        return self.bugname
    @classmethod
    def get_by_bugname(cls,bugname):
        item=db_session.query(BugAdmin).filter(BugAdmin.bugname==bugname).first()
        return item
    @classmethod
    def get_by_id(cls,id):
        item=db_session.query(BugAdmin).filter(BugAdmin.id==id).first()
        return item
    @classmethod
    def get_by_porject_name(cls,porject_name):
        item=db_session.query(BugAdmin).filter(BugAdmin.porject_name==porject_name).first()
        return item
    @classmethod
    def get_count(cls):
        return db_session.query(BugAdmin).count()
class TestCase(Base):
    __tablename__='testcases'
    id=Column(Integer(),primary_key=True)
    porject_id=Column(Integer(),ForeignKey('projects.id'))
    casename=Column(String(64))
    case_qianzhi=Column(String())
    case_buzhou=Column(String())
    case_yuqi=Column(String())
    status=Column(Integer(),default=0)
    case_crea_time=Column(DateTime(),default=datetime.datetime.now())
    user_id=Column(Integer(),ForeignKey('users.id'))
    def __repr__(self):
        return self.casename
    @classmethod
    def get_by_project_name(Cls,project_name):
        item=db_session.query(TestCase).filter(TestCase.project_name==project_name).first()
        return item
    @classmethod
    def get_by_casename(Cls,casename):
        item=db_session.query(TestCase).filter(TestCase.casename==casename).first()
        return item
    @classmethod
    def get_by_id(cls,id):
        item=db_session.query(TestCase).filter(TestCase.id==id).first()
        return item
    @classmethod
    def get_count(cls):
        return db_session.query(TestCase).count()
class BugLog(Base):
    __tablename__='buglogs'
    id=Column(Integer(),primary_key=True)
    bug_id=Column(Integer(),ForeignKey('bugadmins.id'))
    caozuo=Column(String())
    caozuo_time=Column(DateTime())
    user_id=Column(Integer(),ForeignKey('users.id'))
    def __repr__(self):
        return self.caozuo
    @classmethod
    def get_by_id(Cls,id):
        item=db_session.query(BugLog).filter(BugLog.id==id).first()
        return item
    @classmethod
    def get_by_user_id(Cls,user_id):
        item=db_session.query(BugLog).filter(BugLog.user_id==user_id).first()
        return item
    @classmethod
    def get_by_bug_id(Cls,bug_id):
        item=db_session.query(BugLog).filter(BugLog.bug_id==bug_id).first()
        return item
class Project(Base):
    __tablename__='projects'
    id=Column(Integer(),primary_key=True)
    name=Column(String(64))
    user_id=Column(Integer(),ForeignKey('users.id'))
    bug_log=relationship('BugAdmin',backref='projects')
    banben=relationship('BanbenWrite',backref='projects')
    testresult=relationship('TestResult',backref='projects')
    testcase=relationship('TestCase',backref='projects')
    def __repr__(self):
        return self.name
    @classmethod
    def get_by_id(cls,id):
        item=db_session.query(Project).filter(Project.id==id).first()
        return item
    @classmethod
    def get_by_name(cls,name):
        item=db_session.query(Project).filter(Project.name==name).first()
        return item

这是数据库相关的，

数据库配置相关的
from sqlalchemy import  create_engine
from  sqlalchemy.orm import  scoped_session,sessionmaker
from sqlalchemy.ext.declarative import  declarative_base
engine=create_engine('sqlite:///shebei.db',convert_unicode=True)
Base=declarative_base()
db_session=scoped_session(sessionmaker(bind=engine))
def create_all():
    Base.metadata.create_all(engine)
def drop_all():
    Base.metadata.drop_all(engine)

其实在开发的过程中，也遇到了很多阻力，比如下载一直实现不好，比如分页，也是参照别人的实现的，

分页公共模块

class Pagination:
    def __init__(self, current_page, all_item):
        try:
            page = int(current_page)
        except:
            page = 1
        if page < 1:
            page = 1
        all_pager, c = divmod(all_item, 10)
        if int(c) > 0:
            all_pager += 1
        self.current_page = page
        self.all_pager = all_pager
    @property
    def start(self):
        return (self.current_page - 1) * 10
    @property
    def end(self):
        return self.current_page * 10
    def string_pager(self, base_url="/index/"):
        if self.current_page == 1:
            prev = '<li><a href="javascript:void(0);">上一页</a></li>'
        else:
            prev = '<li><a href="%s%s">上一页</a></li>' % (base_url, self.current_page - 1,)
        if self.current_page == self.all_pager:
            nex = '<li><a href="javascript:void(0);">下一页</a></li>'
        else:
            nex = '<li><a href="%s%s">下一页</a></li>' % (base_url, self.current_page + 1,)
        last = '<li><a href="%s%s">尾页</a></li>' % (base_url, self.all_pager,)
        str_page = "".join((prev,nex,last))
        return str_page

在上传文件的时候，原来存放在本地，结果呢，下载处理不好，于是乎选择了七牛，需要到七牛的官网去注册自己的账号

from qiniu import Auth,put_file,etag,urlsafe_base64_encode
import qiniu.config
access_key='uVxowDUcYx641ivtUb111WBEI4112L3D117JHNM_AOtskRh4'
secret_key='PdXU9XrXTLtp1N21bhU1Frm1FDZqE1qhjkEaE9d1xVLZ5C'
def sendfile(key,file):
    q=Auth(access_key,secret_key)
    bucket_name='leilei22'
    token = q.upload_token(bucket_name, key)
    ret, info = put_file(token, key, file)
    me= ret['hash']
    f=etag(file)
    if me==f:
        assert_t=True
    else:
        assert_t=False
    return assert_t

解析Excel，主要用于上传测试用例

import xlrd,xlwt
from xlutils.copy import copy
def datacel(filepath):
    file=xlrd.open_workbook(filepath)
    me=file.sheets()[0]
    nrows=me.nrows
    porject_id_list=[]
    casename_list=[]
    case_qianzhi_list=[]
    case_buzhou_list=[]
    case_yuqi_list=[]
    for i in range(1,nrows):
        porject_id_list.append(me.cell(i,0).value)
        casename_list.append(me.cell(i,2).value)
        case_qianzhi_list.append(me.cell(i,3).value)
        case_buzhou_list.append(me.cell(i,4).value)
        case_yuqi_list.append(me.cell(i,1).value)
    return porject_id_list,casename_list,case_qianzhi_list,case_buzhou_list,case_yuqi_list

其实这么现在公共模块完毕了，其实现在可以着手去开始写我们的代码了，主要的代码，还有静态界面，因为前后端都是我自己，我的前端其实还是从网上找来的模板，

学习的道路是痛苦的，但是我相信我是可以成功的，

from tornado.web import RequestHandler
from models.model_py import User
class BaseHandler(RequestHandler):
    @property
    def db(self):
        return self.application.db
    def get_current_user(self):
        user_id = self.get_secure_cookie('user_id')
        if not user_id:
            return None
        return User.get_by_id(int(user_id))

基础的类集成了RequestHandler的类，，进行了一些简单的自定义，然后后续可以用这个，

进过两周的开发，已经形成了成熟的，


文章标签：
Python
前端开发
测试技术
数据库管理
数据库
存储
关键词：
Python系统
Python类
Python tornado
1032199045570561
目录
相关文章
游客pxynxsrl3ycx6
|
17天前
|
存储 人工智能 搜索推荐
【python】python用户管理系统[简易版]（源码+报告）【独一无二】
【python】python用户管理系统[简易版]（源码+报告）【独一无二】
游客pxynxsrl3ycx6
31 0
实沈
|
22天前
|
Python
Python实现简易天气查询系统
Python实现简易天气查询系统
实沈
26 4
小白学大数据
|
3天前
|
数据采集 NoSQL 搜索推荐
五一假期畅游指南：Python技术构建的热门景点分析系统解读
五一假期畅游指南：Python技术构建的热门景点分析系统解读
小白学大数据
18 0
微芒不朽
|
3天前
|
Python
python面型对象编程进阶（继承、多态、私有化、异常捕获、类属性和类方法）（上）
python面型对象编程进阶（继承、多态、私有化、异常捕获、类属性和类方法）（上）
微芒不朽
37 0
微芒不朽
|
3天前
|
索引 Python
python 格式化、set类型和class类基础知识练习（上）
python 格式化、set类型和class类基础知识练习
微芒不朽
23 0
find_starshine
|
4天前
|
Python
python学习12-类对象和实例对象
python学习12-类对象和实例对象
find_starshine
16 0
米码收割机
|
12天前
|
人工智能 机器人 测试技术
【Python】Python仓储管理系统（源码）【独一无二】
【Python】Python仓储管理系统（源码）【独一无二】
米码收割机
27 0
游客pxynxsrl3ycx6
|
17天前
|
人工智能 机器人 测试技术
【Python】Python房屋销售系统(源码)【独一无二】（课程设计）
【Python】Python房屋销售系统(源码)【独一无二】（课程设计）
游客pxynxsrl3ycx6
29 0
轻松学编程
|
19天前
|
机器学习/深度学习 数据采集 算法
基于Apriori关联规则的电影推荐系统（附python代码）
这是一个基于Apriori算法的电影推荐系统概览。系统通过挖掘用户评分数据来发现关联规则，例如用户观看某部电影后可能感兴趣的其他电影。算法核心是逐层生成频繁项集并设定最小支持度阈值，之后计算规则的置信度。案例中展示了数据预处理、频繁项集生成以及规则提取的过程，具体包括用户评分电影的统计分析，如1-5部电影的评分组合。最后，通过Python代码展示了Apriori算法的实现，生成推荐规则，并给出了一个简单的推荐示例。整个过程旨在提高推荐的精准度，基于用户已评分的电影推测他们可能尚未评分但可能喜欢的电影。
轻松学编程
45 1
基于Apriori关联规则的电影推荐系统（附python代码）
实沈
|
22天前
|
Python
Python实现简易文件管理系统
Python实现简易文件管理系统
实沈
27 5

热门文章

最新文章

  • 1
    【Python 基础教程 02】 数据类型全解析：从基础到高级，实用指南及详细使用案例
  • 2
    【Python 基础教程 01 全面介绍】 Python编程基础全攻略：一文掌握Python语法精髓，从C/C++ 角度学习Python的差异
  • 3
    使用Python实现DBSCAN聚类算法
  • 4
    在Python Web开发过程中：数据库与缓存，MySQL和NoSQL数据库的主要差异是什么？
  • 5
    Redis与Python的完美结合：实现高效数据交互和应用场景全解析
  • 6
    【Python 基础教程 23】Python3 错误与异常处理全面指南：从入门到精通的实用教程
  • 7
    Python读取.nc文件的方法与技术详解
  • 8
    Pandas基本操作：Series和DataFrame（Python）
  • 9
    【Python 基础教程 运算符06】Python3运算符超详细解析：全面入门教程，初学者必读
  • 10
    【Python 基础教程 05】超详细解析Python3注释：全面入门教程，初学者必读,了解Python如何 进行注释
  • 1
    「多线程大杀器」Python并发编程利器：ThreadPoolExecutor，让你一次性轻松开启多个线程，秒杀大量任务！
    31
  • 2
    深入理解Python数据结构中的深浅拷贝
    23
  • 3
    Jinja2：使用Python进行模板渲染的艺术
    52
  • 4
    深入理解Python中的try-except语句，避免代码崩溃的噩梦
    17
  • 5
    python编程简介（一）
    93
  • 6
    python操作列表方法（二）
    29
  • 7
    python操作列表方法（一）
    25
  • 8
    python测试代码（三）
    18
  • 9
    python测试代码（二）
    19
  • 10
    python测试代码（一）
    17

    • 相关课程

    更多
  • Python Web开发基础
  • Python开发基础入门
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 相关实验场景

    更多
  • 用Python画圣诞树
  • 使用Python完成RDS数据开发及分析
  • 语言入门-1：环境构建
  • 阿里云平台上进行Java程序的编译与运行
  • Python选择及循环结构
  • 云端Python及基本操作
    • 下一篇
    阿里云oss简介和使用流程