我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用bottle.template()。
def diff_jobs(): user = root.authorized() app = root.active_app() selected_cases = request.query.selected_diff_cases cases = selected_cases.rstrip(':').split(':') cids = list() contents = list() for jid in cases: cid = jobs(jid).cid cids.append(cid) app = jobs(jid).app base_dir = os.path.join(user_dir, user, root.myapps[app].appname) fn = os.path.join(base_dir, cid, root.myapps[app].simfn) content = slurp_file(fn).splitlines(1) contents.append(content) import difflib d = difflib.Differ() result = list(d.compare(contents[0], contents[1])) title = "diff " + cids[0] + " " + cids[1] params = { 'cid': cid, 'contents': ' '.join(result), 'app': app, 'user': user, 'fn': title } return template('more', params)
def chart(num_bars): """Creates a bar chart with the desired number of bars in a chart.""" if num_bars <= 0: num_bars = 1 data = {"days": [], "bugs": [], "costs": []} for i in range(1, num_bars + 1): data['days'].append(i) data['bugs'].append(random.randint(1,100)) data['costs'].append(random.uniform(1.00, 1000.00)) hover = create_hover_tool() plot = create_bar_chart(data, "Bugs found per day", "days", "bugs", hover) script, div = components(plot) return template(TEMPLATE_STRING, bars_count=num_bars, the_div=div, the_script=script)
def get_main(): services = [] for service in config.sections(): service_status = get_service_action(service, 'status') if service_status['status'] == 'not-found': cls = 'active' elif service_status['status'] == 'inactive' or service_status['status'] == 'failed': cls = 'danger' elif service_status['status'] == 'active': cls = 'success' else: cls = 'warning' disabled_start = True if cls == 'active' or cls == 'success' else False disabled_stop = True if cls == 'active' or cls == 'danger' else False disabled_restart = True if cls == 'active' or cls == 'danger' else False services.append({'class': cls, 'disabled_start': disabled_start, 'disabled_stop': disabled_stop, 'disabled_restart': disabled_restart, 'title': config.get(service, 'title'), 'service': service}) return template('index', hostname=gethostname(), services=services)
def status(): cron = [] task = [] for item in db.cron.fetchall(): plan, status = db.status.fetch(item['id']) last = status[1:status.find(']')] status = status.split(' - ')[-1] cron.append([item['path'], plan, last, status]) for item in db.taskq.fetchall(): length = db.task.length(item[0]) task.append(list(item) + [length]) context = { 'title': '????', 'cron': cron, 'task': task, 'conf': conf, 'notice': flash() } return template(stpl.status, context)
def hello2_name(name): """ ???????????????????????????????????? bottle????????????????????????????????? ????./views/?????????????????????? ??????????????????? ?????????????????????????????? eg. display_name="TEST NAME" :param name: :return: """ return template("sample", display_name=name) # ??????
def chat_room(): """ ????????? :return: """ # cookie??????request???? username = request.get_cookie("username") # cookie??????????????????? if not username: return redirect("/") # ????????????????????? talk_list = get_talk() return template("chat_room", username=username, talk_list=talk_list)
def show_app(app): # very similar to start_new_job() consider consolidating user = root.authorized() root.set_active(app) # parameters for return template if app not in root.myapps: return template('error', err="app %s is not installed" % (app)) try: params = {} params.update(root.myapps[app].params) params['cid'] = '' params['app'] = app params['user'] = user params['apps'] = root.myapps return template(os.path.join('apps', app), params) except: exc_type, exc_value, exc_traceback = sys.exc_info() print traceback.print_exception(exc_type, exc_value, exc_traceback) redirect('/app/'+app)
def showapps(): user = root.authorized() q = request.query.q if not q: result = db().select(apps.ALL, orderby=apps.name) else: result = db(db.apps.name.contains(q, case_sensitive=False) | db.apps.category.contains(q, case_sensitive=False) | db.apps.description.contains(q, case_sensitive=False)).select() # find out what apps have already been activated so that a user can't activate twice uid = users(user=user).id activated = db(app_user.uid == uid).select() activated_apps = [] for row in activated: activated_apps.append(row.appid) if user == "admin": configurable = True else: configurable = False params = { 'configurable': configurable, 'user': user } return template('apps', params, rows=result, activated=activated_apps)
def delete_app(appid): user = root.authorized() if user != 'admin': return template('error', err="must be admin to edit app") appname = request.forms.app del_app_dir = request.forms.del_app_dir try: if user == 'admin': # delete entry in DB if del_app_dir == "on": del_files = True else: del_files = False root.myapps[appname].delete(appid, del_files) else: return template("error", err="must be admin") except: exc_type, exc_value, exc_traceback = sys.exc_info() print traceback.print_exception(exc_type, exc_value, exc_traceback) return template("error", err="failed to delete app... did the app load properly?") redirect("/apps")
def addapp(): user = root.authorized() if user != 'admin': return template('error', err="must be admin to add app") appname = request.forms.appname input_format = request.forms.input_format # ask for app name category = request.forms.category language = request.forms.language description = request.forms.description command = request.forms.command preprocess = request.forms.preprocess postprocess = request.forms.postprocess # put in db a = apprw.App() #print "user:",user a.create(appname, description, category, language, input_format, command, preprocess, postprocess) # load_apps() needs to be called here in case a user wants to delete # this app just after it has been created... it is called again after # the user uploads a sample input file root.load_apps() redirect('/app/'+appname)
def change_password(): # this is basically the same coding as the register function # needs to be DRY'ed out in the future user = root.authorized() if config.auth and not root.authorized(): redirect('/login') opasswd = request.forms.opasswd pw1 = request.forms.npasswd1 pw2 = request.forms.npasswd2 # check old passwd #user = request.forms.user if _check_user_passwd(user, opasswd) and pw1 == pw2 and len(pw1) > 0: u = users(user=user) u.update_record(passwd=_hash_pass(pw1)) db.commit() else: return template('error', err="problem with password") params = {} params['user'] = user params['alert'] = "SUCCESS: password changed" return template('account', params)
def get_docker(): user = root.authorized() params = {} try: cli = docker.Client(base_url=base_url) images = cli.images() conts = cli.containers(all=True) except: images = [] conts = [] params['alert'] = "ERROR: there was a problem talking to the Docker daemon..." params['user'] = user params['app'] = root.active_app() if request.query.alert: params['alert'] = request.query.alert return template('docker', params, images=images, containers=conts)
def get_all_jobs(): user = root.authorized() if not user == "admin": return template("error", err="must be admin to use this feature") cid = request.query.cid app = request.query.app or root.active_app() n = request.query.n if not n: n = config.jobs_num_rows else: n = int(n) # sort by descending order of jobs.id result = db((db.jobs.uid==users.id)).select(orderby=~jobs.id)[:n] # clear notifications users(user=user).update_record(new_shared_jobs=0) db.commit() params = {} params['cid'] = cid params['app'] = app params['user'] = user params['n'] = n params['num_rows'] = config.jobs_num_rows return template('shared', params, rows=result)
def get_aws(): user = root.authorized() cid = request.query.cid app = request.query.app or root.active_app() uid = db(users.user==user).select(users.id).first() #creds = db().select(db.aws_creds.ALL) creds = db(aws_creds.uid==uid).select() # look for aws instances registered by the current user # which means first need to get the uid instances = db(aws_instances.uid==uid).select() params = {} params['cid'] = cid params['app'] = app params['user'] = user if request.query.status: params['status'] = request.query.status return template('aws', params, creds=creds, instances=instances)
def get_user_data(filepath): user = root.authorized() # filepath = request.query.filepath # get the owner from the filepath # e.g. "user_data/wes/mendel/y23022/file.dat" path_list = filepath.split("/") owner = path_list[0] cid = path_list[2] shared = jobs(cid=cid).shared # print filepath, path_list, shared # only allow admin to see other user's cases that have not been shared if owner != user and shared != "True" and user != "admin": return template('error', err="access forbidden") return static_file(filepath, root=user_dir)
def admin_delete_user(): user = root.authorized() if not user == "admin": return template("error", err="must be admin to delete") uid = request.forms.uid if int(uid) == 0: return template("error", err="can't delete admin user") if request.forms.del_files == "True": path = os.path.join(user_dir, users(uid).user) print "deleting files in path:", path if os.path.isdir(path): shutil.rmtree(path) del db.users[uid] db.commit() redirect("/admin/show_users")
def get_stats(): root.authorized() params = {} # number of jobs in queued, running, and completed states params['nq'] = db(jobs.state=='Q').count() params['nr'] = db(jobs.state=='R').count() params['nc'] = db(jobs.state=='C').count() params['cpu'] = psutil.cpu_percent() params['vm'] = psutil.virtual_memory().percent params['disk'] = psutil.disk_usage('/').percent params['cid'] = request.query.cid params['app'] = request.query.app return template("stats", params)
def main_form(): """Main page""" global saved_profile drivers = collection.get_families() if not saved_profile: saved_profile = request.get_cookie('indiserver_profile') or 'Simulators' profiles = db.get_profiles() return template(os.path.join(views_path, 'form.tpl'), profiles=profiles, drivers=drivers, saved_profile=saved_profile) ############################################################################### # Profile endpoints ###############################################################################
def edit_item(no): if request.GET.get('save','').strip(): edit = request.GET.get('task','').strip() status = request.GET.get('status','').strip() if status == 'open': status = 1 else: status = 0 c = conn.cursor() c.execute("UPDATE todo SET task = ?, status = ? WHERE id LIKE ?" ,(edit, status, no)) conn.commit() output = template("tpl/notice",msg = '??????' , no=no) return output else: c = conn.cursor() c.execute("SELECT task FROM todo WHERE id LIKE ?" ,[str(no)]) cur_data = c.fetchone() return template('tpl/edit_task', old=cur_data, no=no)
def edit_item(no): if request.GET.get('save','').strip(): name_r = request.GET.get('name', '').strip() comment_r = request.GET.get('comment', '').strip() c = conn.cursor() c.execute("UPDATE comments SET name = ?, comment = ? WHERE id LIKE ?" ,(name_r, comment_r, no)) conn.commit() return "<script>window.location.replace('/');</script>" else: c = conn.cursor() c.execute("SELECT name, comment FROM comments WHERE id LIKE ?" ,[str(no)]) cur_data = c.fetchone() #print cur_data return template('tpl/edit', old=cur_data, no=no) # ---------------------------------------------------- # ????
def cmd_toolkit(): cl_statement = request.forms.get('cl') # xmlservice itool = iToolKit() itransport = iLibCall() itool.add(iCmd5250(cl_statement, cl_statement)) itool.call(itransport) # results from list data = '' for output_outer in itool.list_out(): for output_inner in output_outer: data += output_inner return template('cmd', data=data)
def edit(): #file_list = { # 'filename1': 'path1', # 'filename2': 'path2', # 'dirname1': { # 'filename3': 'path3', # 'dirname2': { # 'filename4': 'path4', # 'filename5': 'path5' # } # } #} file_list = make_file_tree(ROOT) file = request.GET.get('file') if file: with open(os.path.join(ROOT, file), 'r') as in_file: code = in_file.read() if file.split('.')[-1] in ['pyui', 'json']: code = json.dumps(json.loads(code), indent=4, separators=(',', ': ')) output = template(os.path.join(IDE_REL_ROOT, 'main.tpl'), files = file_list, save_as = file, code = code) else: output = template(os.path.join(IDE_REL_ROOT, 'main.tpl'), files = file_list) return output
def unsub_email(email, code): for user in r.table('users').get_all(["email", email], index = 'identity_check').run(conn()): if user['authlink'] == code: user['subscribed'] = False r.table('users').get_all(["email", email], index = 'identity_check').replace(user).run(conn()) users.flash('info', 'You\'ve been unsubscribed from further emails.') users.login(user) return # for xenforo users for user in r.table('users').filter({'email': email}).run(conn()): if user['authlink'] == code: user['subscribed'] = False r.table('users').get_all(["xenforo", user['identity']]).replace(user).run(conn()) users.flash('info', 'You\'ve been unsubscribed from further emails.') users.login(user) return users.flash('error', 'Authorization failed') return template("login.htm")
def do_login(): username = request.forms.get('username') password = get_salted_password() current_user = user.get_by_username(username) logined = current_user and current_user.salted_password == password if logined: response.set_cookie('ssl_uid', str(current_user.id)) response.set_cookie('ssl_pw', password) return redirect('/') return template('login', username=username, message='User not found.' if not current_user else 'Password is incorrect.', salt=config.USER_SALT )
def new_item(): if request.GET.get('save','').strip(): #????? new=request.GET.get('task','').strip() conn=sqlite3.connect('todo.db') c=conn.cursor() c.execute('INSERT INTO todo (task,status) VALUES (?,?)',(new,1)) new_id=c.lastrowid conn.commit() c.close() return "<p> the new task was inserted into the database,the id is %s</p>" % new_id else: return template('new_task.html')
def edit_item(num): if request.GET.get('save'): edit=request.GET.get('task').strip() status=request.GET.get('status').strip() if status=='open': status=1 else: status=0 conn=sqlite3.connect('todo.db') c=conn.cursor() c.execute('UPDATE todo SET task=?,status=? WHERE id LIKE ?',(edit,status,num)) conn.commit() return "<p>the item number %s was successfully update" % num else: conn = sqlite3.connect('todo.db') c = conn.cursor() c.execute('SELECT task FROM todo WHERE id LIKE ?',(str(num))) cur_data=c.fetchone() return template('edit_task.html',old=cur_data,num=num)
def show_post(permalink="notfound"): cookie = bottle.request.get_cookie("session") username = sessions.get_username(cookie) permalink = cgi.escape(permalink) print "about to query on permalink = ", permalink post = posts.get_post_by_permalink(permalink) if post is None: bottle.redirect("/post_not_found") # init comment form fields for additional comment comment = {'name': "", 'body': "", 'email': ""} return bottle.template("entry_template", dict(post=post, username=username, errors="", comment=comment)) # used to process a comment on a blog post
def process_signup(): email = bottle.request.forms.get("email") username = bottle.request.forms.get("username") password = bottle.request.forms.get("password") verify = bottle.request.forms.get("verify") # set these up in case we have an error case errors = {'username': cgi.escape(username), 'email': cgi.escape(email)} if validate_signup(username, password, verify, email, errors): if not users.add_user(username, password, email): # this was a duplicate errors['username_error'] = "Username already in use. Please choose another" return bottle.template("signup", errors) session_id = sessions.start_session(username) print session_id bottle.response.set_cookie("session", session_id) bottle.redirect("/welcome") else: print "user did not validate" return bottle.template("signup", errors)
def present_welcome(): # check for a cookie, if present, then extract value cookie = bottle.request.get_cookie("session") username = sessions.get_username(cookie) # see if user is logged in if username is None: print "welcome: can't identify user...redirecting to signup" bottle.redirect("/signup") return bottle.template("welcome", {'username': username}) # Helper Functions #extracts the tag from the tags form element. an experience python programmer could do this in fewer lines, no doubt
def show_post(permalink="notfound"): cookie = bottle.request.get_cookie("session") username = sessions.get_username(cookie) permalink = cgi.escape(permalink) print "about to query on permalink = ", permalink post = posts.get_post_by_permalink(permalink) if post is None: bottle.redirect("/post_not_found") # init comment form fields for additional comment comment = {'name': "", 'body': "", 'email': ""} return bottle.template("entry_template", {"post": post, "username": username, "errors": "", "comment": comment}) # used to process a comment on a blog post
def process_signup(): email = bottle.request.forms.get("email") username = bottle.request.forms.get("username") password = bottle.request.forms.get("password") verify = bottle.request.forms.get("verify") # set these up in case we have an error case errors = {'username': cgi.escape(username), 'email': cgi.escape(email)} if validate_signup(username, password, verify, email, errors): if not users.add_user(username, password, email): # this was a duplicate errors['username_error'] = ("Username already in use. " + "Please choose another") return bottle.template("signup", errors) session_id = sessions.start_session(username) print session_id bottle.response.set_cookie("session", session_id) bottle.redirect("/welcome") else: print "user did not validate" return bottle.template("signup", errors)
def present_welcome(): # check for a cookie, if present, then extract value cookie = bottle.request.get_cookie("session") username = sessions.get_username(cookie) # see if user is logged in if username is None: print "welcome: can't identify user...redirecting to signup" bottle.redirect("/signup") return bottle.template("welcome", {'username': username}) # Helper Functions # extracts the tag from the tags form element. An experienced python # programmer could do this in fewer lines, no doubt
def index_tpl(**kwargs): return template('index', **kwargs)
def show_main_page(user=None): user = user or get_logged_in_user() if user is None: return template('login', null=None) heading = 'Posts from people you follow' posts = pubsub.posts_for_user(user) return dict(user=user, posts=posts, heading=heading, comb=comb)
def post_message(): user = get_logged_in_user() if user is None: return template('login', null=None) text = request.forms.get('text', '') if text: pubsub.post_message(user, text) return show_main_page(user)
def prepare(self, **options): from mako.template import Template from mako.lookup import TemplateLookup from plim import preprocessor options.update({'input_encoding':self.encoding}) options.setdefault('format_exceptions', bool(DEBUG)) lookup = TemplateLookup(directories=self.lookup, **options) if self.source: self.tpl = Template(self.source, preprocessor = preprocessor, lookup=lookup, **options) else: self.tpl = Template(uri=self.name, preprocessor = preprocessor, filename=self.filename, lookup=lookup, **options)
def stpl(tsk): ps = tsk.inputs[0].abspath() pt = tsk.outputs[0].abspath() bld = tsk.generator.bld lookup,name=os.path.split(ps) st=bottle.template(name,template_lookup=[lookup], company = bld.env.company, guiname=bld.env.guiname, version=bld.env.version, dllname=bld.env.dllname, maxfuni=bld.env.maxfuni) with codecs.open(pt,mode='w',encoding="utf-8") as f: f.write(st) os.chmod(pt, 493) # copy files that already exist
def show_message(msg): """Display a message if the msg value is greater than 2 characters in the path. """ valid_length = len(msg) >= MIN_MSG_LENGTH valid_name = re.match('^[a-z\-]+$', msg.lower()) is not None if valid_length and valid_name: return template(TEMPLATE_STRING, h1=msg) else: error_msg = "Sorry, only alpha characters and hyphens allowed." raise Exception(error_msg)
def get_service_journal_page(service): if service in config.sections(): if get_service_action(service, 'status')['status'] == 'not-found': abort(400,'Sorry, but service \'{}\' unit not found in system.'.format(config.get(service, 'title'))) journal_lines = get_service_journal(service, 100) return template('journal', hostname=gethostname(), service=config.get(service, 'title'), journal=journal_lines['journal']) else: abort(400, 'Sorry, but \'{}\' is not defined in config.'.format(service)) # Serve static content
def homepage(): return template(stpl.homepage, {'conf': conf})
def reload_cron(): worker.load_crontab() previous = request.headers.get('Referer', '/') return template(stpl.reload_cron, previous=previous)
def log(): mode = request.query.get('mode', 'cron') sort = request.query.get('sort', 'new') page = int(request.query.page or 1) data = [] if os.path.exists(conf.log): with open(conf.log, 'rb') as f: lines = map(lambda s: s.decode('utf8'), f.readlines()) data = [line for line in lines if line.startswith(mode.title())] data = data[::-1] if sort == 'new' else data neg_sort = { 'new': {'sort': 'old', 'title': '????'}, 'old': {'sort': 'new', 'title': '????'} } neg_mode = { 'cron': {'mode': 'task', 'title': '????'}, 'task': {'mode': 'cron', 'title': '????'} } context = { 'title': '?? %s ??' % mode.title(), 'data': data[(page - 1) * 10: page * 10], 'mode': mode, 'page': page, 'count': len(data), 'sort': neg_sort[sort], 'other': neg_mode[mode] } return template(stpl.log, context)
def index(): return template("index")
def showmyapps(): user = root.authorized() uid = users(user=user).id app = root.active_app() result = db((apps.id == app_user.appid) & (uid == app_user.uid)).select(orderby=apps.name) if user == "admin": configurable = True else: configurable = False params = { 'configurable': configurable, 'user': user, 'app': app } return template('myapps', params, rows=result)
def app_edit(appid): user = root.authorized() if user != 'admin': return template('error', err="must be admin to edit app") cid = request.forms.cid app = request.forms.app result = db(apps.name==app).select().first() params = {'app': app, 'cid': cid} return template('app_edit', params, rows=result)
def getaddapp(): user = root.authorized() if user != 'admin': return template('error', err="must be admin to add app") return template('appconfig/addapp')
def appconfig_status(): root.authorized() status = dict() app = request.query.app # check db file appname = apps(name=app).name if appname: status['command'] = 1 else: status['command'] = 0 # check template file if os.path.exists("views/apps/"+app+".tpl"): status['template'] = 1 else: status['template'] = 0 # check inputs file extension = {'namelist':'.in', 'ini':'.ini', 'xml':'.xml', 'json':'.json', 'yaml':'.yaml', 'toml':'.toml'} if os.path.exists(os.path.join(apprw.apps_dir, app, app + extension[root.myapps[app].input_format])): status['inputs'] = 1 else: status['inputs'] = 0 # check app binary if os.path.exists(os.path.join(apprw.apps_dir, app, app)): status['binary'] = 1 else: status['binary'] = 0 # check plots appid = apps(name=app).id result = db(plots.appid==appid).select().first() if result: status['plots'] = 1 else: status['plots'] = 0 return json.dumps(status)
def appconfig_exe(step="upload"): user = root.authorized() if user != 'admin': return template('error', err="must be admin to configure app") if step == "upload": appname = request.forms.appname params = {'appname': appname} return template('appconfig/exe_upload', params) elif step == "test": appname = request.forms.appname upload = request.files.upload if not upload: return template('appconfig/error', err="no file selected. press back button and try again") name, ext = os.path.splitext(upload.filename) # if ext not in ('.exe','.sh','.xml','.json',): # return 'ERROR: File extension apps not allowed.' try: save_path_dir = os.path.join(apprw.apps_dir, name) if not os.path.exists(save_path_dir): os.makedirs(save_path_dir) save_path = os.path.join(save_path_dir, name) + ext if os.path.isfile(save_path): timestr = time.strftime("%Y%m%d-%H%M%S") shutil.move(save_path, save_path+"."+timestr) upload.save(save_path) os.chmod(save_path, 0700) # process = subprocess.Popen(["otool -L", save_path], stdout=subprocess.PIPE, stdin=subprocess.PIPE) # contents = process.readlines() contents = "SUCCESS" params = {'appname': appname, 'contents': contents} return template('appconfig/exe_test', params) except IOError: exc_type, exc_value, exc_traceback = sys.exc_info() print traceback.print_exception(exc_type, exc_value, exc_traceback) return "IOerror:", IOError else: return "ERROR: must be already a file"
def get_register(): return template('register')
