最近樓主加班 嘍, 好久沒有更新我的博客了,哎,一言難盡,廢話我就不說了,來開始上精華。 背景:目前市面上有很多bug管理工具,但是各有各的特點,最著名,最流行的就是禪道,一個偶然的機會接觸到了python ,學到tornado後,就想著去怎麼去用到實處,後來發現自己公司的除了禪道就記錄bug沒有什 ...
最近樓主加班 嘍, 好久沒有更新我的博客了,哎,一言難盡,廢話我就不說了,來開始上精華。
背景:目前市面上有很多bug管理工具,但是各有各的特點,最著名,最流行的就是禪道,一個偶然的機會接觸到了python ,學到tornado後,就想著去怎麼去用到實處,後來發現自己公司的除了禪道就記錄bug沒有什麼可以用的工具了。
語言:python3 第三庫 :tornado,qiniu(用於雲存儲文件),資料庫用sqlite
why use tornado?很多人其實會這麼問我,我感覺tornado可以實現非同步,雖然現在代碼還沒有用到非同步,我感覺還是很不錯的框架,值得學習,現在很多公司都在用,個人感覺這是一個不錯的,值得我們大家去學習的框架。
來看看我的需求文檔
大題是這麼的輪廓,那麼拿到這個的時候,我會進行需求分析,
需要什麼樣的資料庫, 數據模型之間的關係,雖然現在還是有很多地方是寫死的還沒有進行搜索功能的設置,但是我相信,有了現在這個demo
那麼我開始來設計下我主要會結構
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
這是資料庫相關的,
1 資料庫配置相關的 2 3 from sqlalchemy import create_engine 4 from sqlalchemy.orm import scoped_session,sessionmaker 5 from sqlalchemy.ext.declarative import declarative_base 6 engine=create_engine('sqlite:///shebei.db',convert_unicode=True) 7 Base=declarative_base() 8 db_session=scoped_session(sessionmaker(bind=engine)) 9 def create_all(): 10 Base.metadata.create_all(engine) 11 def drop_all(): 12 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的類,,進行了一些簡單的自定義,然後後續可以用這個,
進過兩周的開發,已經形成了成熟的,
源碼已經開源道github
https://github.com/liwanlei/tornado_guanli
另外本人提供vip指導,python自動化,python學習上面,python做測試開發上面,我都會提供指導,vip終身有效,現價700,聯繫qq:952943386,稍後會更新qq群,喜歡的朋友可以來,本人也是從一個不懂啥是python到現在能做東西,從15的2k到17的10k+都是辛酸淚,可以與大家一起共勉,提供思路,能夠讓你走的更遠,我也是從白紙一張一步步走來,如果你是新畢業的學生這個也適合你,可以給你指導,但是我能做的就是帶你,但是我不是培訓班,沒有提供就業的機會,我只會在有機會的時候給你推薦,學無止境,此時可能我只能教你一些測試,提供職業指導,我可能以後會帶給你更多。
我是一個大專生,也是一個人奮鬥到今天。