Python app.app 模块，models() 实例源码

def ping(service_id):

    from app import db, models
    from models import Service
    finding = Service.query.filter_by(serviceid=service_id).first()
    if finding is not None:
        image_name = finding.imagename
        uploadn = finding.uploadname
        usern = finding.username
        firstcreatetime = finding.firstcreatetime
        u = Service(serviceid = service_id, createdtime = str(time.time()), imagename = image_name, uploadname = uploadn, username = usern, firstcreatetime = firstcreatetime)
        db.session.add(u) 
        db.session.commit() 
        db.session.delete(finding)
        db.session.commit()
    else:
        return "The service "+service_id+" has been removed!"

    return "There are existing service:"+service_id

def dispatch_request(self):
        if not  session.get('username'):
            return  redirect(url_for('home.login'))
        if request.method=="POST":
            model=request.form.get('project')
            if model=='':
                flash(u'????????')
                return render_template('add/add_moel.html')
            user_id=User.query.filter_by(username=session.get('username')).first().id
            models=Model.query.filter_by(model_name=model).first()
            if models:
                flash(u'??????')
                return render_template('add/add_moel.html')
            new_moel=Model(model_name=model,model_user_id=user_id)
            db.session.add(new_moel)
            db.session.commit()
            flash(u'????!')
            return  redirect(url_for('home.model'))
        return  render_template('add/add_moel.html')

def dispatch_request(self,id):
        if not session.get('username'):
            return  redirect(url_for('home.login'))
        user = User.query.filter_by(username=session.get('username')).first()
        model=Model.query.filter_by(id=id).first()
        if request.method=="POST":
            ed_mode=request.form.get('model')
            if ed_mode=='':
                flash(u'??????')
                return render_template('edit/edit_model.html', mode=model)
            models = Model.query.filter_by(model_name=ed_mode).first()
            model.model_name=ed_mode
            try:
                db.session.commit()
                flash(u'????')
                return  redirect(url_for('home.model'))
            except:
                db.session.rollback()
                flash(u'??zhi????')
                return redirect(url_for('home.model'))
        return  render_template('edit/edit_model.html', mode=model)

def dispatch_request(self,id):
        if not session.get('username'):
            return  redirect(url_for('home.login'))
        user = User.query.filter_by(username=session.get('username')).first()
        project=Project.query.filter_by(id=id).first()
        if request.method=="POST":
            ed_mode=request.form.get('project')
            if ed_mode=='':
                flash(u'?????')
                return render_template('edit/edit_pro.html', project=project)
            models = Project.query.filter_by(project_name=ed_mode).first()
            project.project_name=ed_mode
            try:
                db.session.commit()
                flash(u'????')
                return  redirect(url_for('home.project'))
            except:
                db.session.rollback()
                flash(u'???????')
                return redirect(url_for('home.project'))
        return  render_template('edit/edit_pro.html', project=project)

def meraki_redirect():
    phonenumber = request.form.get('phonenumber')
    macaddr = request.form.get('macaddr')
    retype = request.form.get('type')
    redirect_url = request.form.get('redirect_url')
    guest = models.Guest(phone=phonenumber, guest_mac=macaddr)
    db.session.add(guest)
    db.session.commit()
    robj = models.Request(guest=guest, rtype_id=retype, status_id=1)
    db.session.add(robj)
    db.session.commit()
    granted_msg = "Position : " + str(robj.id) + " - " + "wait.thekey.pw/guest/" + phonenumber
    # granted_msg += " | Pour traquer la file d'attente en temps reel : "
    # granted_msg += "http://wait.thekey.pw/guest/" + phonenumber
    # ins = db.session.query(models.Guest).insert().values(phone=phonenumber, macaddress=macaddr, rtype=retype)
    # return jsonify({'url': redirect_url})
    print granted_msg
    sms(phonenumber, granted_msg)
    return redirect(redirect_url, code=302)

def setting():
    from app.forms import SetForm

    form = SetForm()
    if form.validate_on_submit():
        servers = models.ServerIP.query.all()
        for server in servers:
            db.session.delete(server)
            db.session.commit()
        serverip = form.ip.data
        u = models.ServerIP(serverip = serverip)
        db.session.add(u)
        db.session.commit()
        return render_template('setting.html',form=form, succeed = True)
    return render_template('setting.html',form=form)

def images():
    from app import db, models         
    images = models.Image.query.all()
    result = []
    part_line = {'imagename':'default','uploadname':'default','uploaduser':'default','comments':'default'}
    for i in images:
        part_line['imagename'] = i.imagename
        part_line['uploadname'] = i.uploadname
        part_line['uploaduser'] = i.uploaduser
        part_line['comments'] = i.comments
        result.append(part_line)
        part_line = {}
    return render_template('images.html',imagetables = result)

def idetailed(image_name):
    from app import db, models 

    image = models.Image.query.filter_by(imagename = image_name).first()
    published_topics = StringToListOfDict(image.published_topics)
    subscribed_topics = StringToListOfDict(image.subscribed_topics)
    pub_topic_list = []
    sub_topic_list = []
    for pub_topic in published_topics:
        pub_topic_list.append(pub_topic.get("topic_name"))
    for sub_topic in subscribed_topics:
        sub_topic_list.append(sub_topic.get("topic_name"))

    return render_template('idetailed.html',imagename = image.imagename, uploadname = image.uploadname, uploaduser = image.uploaduser, uploadtime = image.uploadtime, subscribed_topics = sub_topic_list, published_topics = pub_topic_list, advertised_services = StringToList(image.advertised_services), advertised_actions = StringToList(image.advertised_actions), comments = image.comments)

def get(self):
        form=InterForm()
        project,models=get_pro_mo()
        return render_template('add/add_interface.html', form=form, projects=project, models=models)

def post(self):
        form=InterForm()
        project,models=get_pro_mo()
        if form.validate_on_submit and request.method =="POST":
            project_name=request.form.get('project')
            model_name=request.form.get('model')
            interface_name=request.form.get('interface_name')
            interface_url=request.form.get('interface_url')
            interface_header=request.form.get('interface_headers')
            interface_meth=request.form.get('interface_meth')
            interface_par=request.form.get('interface_par')
            interface_bas=request.form.get('interface_bas')
            if project_name == None or model_name ==None or interface_header=='' or interface_name=='' or interface_url =='' or interface_meth=='':
                flash(u'????????????')
                return render_template('add/add_interface.html', form=form, projects=project, models=models)
            user_id=User.query.filter_by(username=session.get('username')).first().id
            project_id=Project.query.filter_by(project_name=project_name).first().id
            models_id=Model.query.filter_by(model_name=model_name).first().id
            try:
                new_interface=Interface(model_id=models_id,projects_id=project_id,Interface_name=interface_name,Interface_url=interface_url,Interface_meth=interface_meth,Interface_par=interface_par,Interface_back=interface_bas,Interface_user_id=user_id,Interface_headers=interface_header)
                db.session.add(new_interface)
                db.session.commit()
                flash(u'????')
                return redirect(url_for('home.interface'))
            except:
                db.session.rollback()
                flash(u'????')
                return render_template('add/add_interface.html', form=form, projects=project, models=models)
        return render_template('add/add_interface.html', form=form, projects=project, models=models)

def get(self,id):
        interface=Interface.query.filter_by(id=id).first()
        project, models = get_pro_mo()
        return render_template('edit/edit_inter.html', interface=interface, projects=project, models=models)

def post(self,id):
        interface=Interface.query.filter_by(id=id).first()
        project, models = get_pro_mo()
        if request.method=='POST':
            projecct=request.form.get('project')
            model=request.form.get('model')
            intername=request.form.get('inter_name')
            url=request.form.get('url')
            headers=request.form.get('headers')
            meth=request.form.get('meth')
            reques=request.form.get('reque')
            back=request.form.get('back')
            if projecct is None or model is None or intername=='' or headers =='' or url=='' or meth=='' or back=='':
                flash(u'????????????')
                return render_template('edit/edit_inter.html', interface=interface, projects=project, models=models)
            project_id = Project.query.filter_by(project_name=projecct).first().id
            models_id = Model.query.filter_by(model_name=model).first().id
            interface.projects_id=project_id
            interface.model_id=models_id
            interface.Interface_name=intername
            interface.Interface_headers=headers
            interface.Interface_url=url
            interface.Interface_meth=meth
            interface.Interface_par=reques
            interface.Interface_back=back
            interface.Interface_user_id=User.query.filter_by(username=session.get('username')).first().id
            db.session.commit()
            return redirect(url_for('home.interface'))
        return render_template('edit/edit_inter.html', interface=interface, projects=project, models=models)

def dispatch_request(self):
        form=Interface_yong_Form()
        project, models = get_pro_mo()
        if request.method=='POST' and form.validate_on_submit :
            yongli_nam=request.form.get('project')
            mode=request.form.get('model')
            interface_name=request.form.get('interface_name')
            interface_url=request.form.get('interface_url')
            interface_header=request.form.get('interface_headers')
            interface_meth=request.form.get('interface_meth')
            interface_can=request.form.get('interface_can')
            interface_re=request.form.get('interface_rest')
            if yongli_nam is None or mode is None or interface_name=='' or interface_header==''or interface_url=='' or interface_meth=='' or interface_re=='':
                flash(u'???????')
                return render_template('add/add_test_case.html', form=form, projects=project, models=models)
            project_id = Project.query.filter_by(project_name=yongli_nam).first().id
            models_id = Model.query.filter_by(model_name=mode).first().id
            try:
                newcase=InterfaceTest(projects_id=project_id,model_id=models_id,Interface_name=interface_name,Interface_headers=interface_header,Interface_url=interface_url,Interface_meth=interface_meth,Interface_pase=interface_can,Interface_assert=interface_re,Interface_user_id=User.query.filter_by(username=session.get('username')).first().id)
                db.session.add(newcase)
                db.session.commit()
                flash(u'??????')
                return redirect(url_for('home.yongli'))
            except:
                db.session.rollback()
                flash(u'??????')
                return render_template('add/add_test_case.html', form=form, projects=project, models=models)
        return render_template('add/add_test_case.html', form=form, projects=project, models=models)

def dispatch_request(self):
        project=Project.query.all()
        models=Model.query.all()
        if request.method=='POST':
            projecct=request.form.get('project')
            model=request.form.get('model')
            if projecct =='':
                interd=InterfaceTest.query.filter(InterfaceTest.model_id==int(model)).all()
                return render_template('home/ser_yonglo.html', yonglis=interd, projects=project, models=models)
            if model =='':
                interd=InterfaceTest.query.filter(InterfaceTest.projects_id==int(projecct)).all()
                return render_template('home/ser_yonglo.html', yonglis=interd, projects=project, models=models)
            interd=InterfaceTest.query.filter(InterfaceTest.projects_id==int(projecct),InterfaceTest.model_id==int(model)).order_by('-id').all()
            return render_template('home/ser_yonglo.html', yonglis=interd, projects=project, models=models)
        return redirect(url_for('home.yongli'))

def get(self,id):
        project,models=get_pro_mo()
        event=Interfacehuan.query.filter_by(id=id).first()
        return  render_template('edit/edit_events.html', enents=event, projects=project)

def migrate():
    """Runs database migrations."""
    from app.models import Base
    metadata = Base.metadata
    engine = app.config.get("SQLALCHEMY_ENGINE")
    metadata.create_all(engine)

def shell():
    """Run shell"""
    subprocess.call(['wsgicli', 'shell', 'manage.py', 'app', '-i', '--models', 'ipython'])