我们从Python开源项目中，提取了以下17个代码示例，用于说明如何使用web.database()。
def get_searched_term_ids(usr_in): term_names_db = web.database(dbn='sqlite', db='static/term_names.sqlite') usr_in = usr_in.strip() results = term_names_db.query( "SELECT term_id FROM term_names WHERE term_name=$term_name", vars={'term_name':usr_in} ) term_ids = Set([r["term_id"].encode('utf-8') for r in results]) print "Found term IDs for query '%s': %s" % (usr_in, term_ids) # If no terms were found check if the input was an ontology term ID if len(term_ids) == 0 and len(usr_in.split(":")) == 2: pref = usr_in.split(":")[0] suff = usr_in.split(":")[1] valid_pref = pref in Set(["DOID", "UBERON", "CVCL", "CL", "EFO"]) valid_suff = suff.isdigit() if valid_pref and valid_suff: print "User input '%s' can be interpreted as a term ID" % usr_in term_ids = Set([usr_in]) return term_ids
def get_dbn(cfg_file = "config.cfg"): parser = ConfigParser.SafeConfigParser() parser.optionxform = str cn_db_file = os.getenv("CN_DB") if cn_db_file is not None: print "*** USING DATABASE", cn_db_file cfg_file = cn_db_file parser.read(cfg_file) section = 'database' if 'database' not in parser.sections(): raise Exception("No database section in " + cfg_file) return parser.get(section, 'dbn') #-----------------------------------------------------------------------
def __init__(self): self.db = web.database(dbn='oracle', user='cap', pw='cap_1234', db='oracapdb') self.id = ID() self.time = Time() self.logger = logging.getLogger('cmdbapi')
def connetions(self, dbname, database, user, passwd): if dbname not in self.dblist: print 'you are wrong' return None try: self.dbsocket = web.database(dbn= dbname, user=user, pw=passwd, db=database) except Exception: return Exception return self.dbsocket
def __init__(self): self.db = web.database(dbn='oracle', user='cap', pw='cap_1234', db='oracapdb') self.logger = logging.getLogger('cmdbapi')
def get_dbn(cfg_file="config.cfg"): parser = ConfigParser.SafeConfigParser() parser.optionxform = str parser.read(cfg_file) section = 'database' if 'database' not in parser.sections(): raise Exception("No database section in %s" % cfg_file) return parser.get(section, 'dbn') #-------------------------------------------------------------------------------
def init_web_db(cfg_file="config.cfg"): parser = ConfigParser.SafeConfigParser() parser.optionxform = str parser.read(cfg_file) section = 'database' if 'database' not in parser.sections(): raise Exception("No database section in %s" % cfg_file) dbn = parser.get(section, 'dbn') if dbn == "mysql": db = parser.get(section, 'db') user = parser.get(section, 'user') pw = parser.get(section, 'pw') host = parser.get(section, 'host') db = web.database(dbn='mysql', db=db, user=user, pw=pw, host=host) db.query('SET NAMES utf8;') db.query('SET CHARACTER SET utf8;') db.query('SET character_set_connection=utf8;') elif dbn == "sqlite": dbname = parser.get(section, 'db') db = web.database(dbn='sqlite', db=dbname) # We need to mimic some MySQL functions in order to be able to use # SQLite or use different SQL commands for each database server. I # prefer the 1st option, naturally... db._db_cursor().connection.create_function("concat", 2, sqlite_concat) db._db_cursor().connection.create_function("conv", 3, sqlite_conv) db._db_cursor().connection.create_function("instr", 2, sqlite_instr) db._db_cursor().connection.create_function("rand", 0, sqlite_rand) db._db_cursor().connection.text_factory = str return db #-------------------------------------------------------------------------------
def setup_database(): if web.config.get('db_parameters'): db = web.database(**web.config.db_parameters) web.insert = db.insert web.select = db.select web.update = db.update web.delete = db.delete web.query = db.query def transact(): t = db.transaction() web.ctx.setdefault('transaction_stack', []).append(t) def rollback(): stack = web.ctx.get('transaction_stack') t = stack and stack.pop() t and t.rollback() def commit(): stack = web.ctx.get('transaction_stack') t = stack and stack.pop() t and t.commit() web.transact = transact web.rollback = rollback web.commit = commit
def POST(self): form = search_form() form.validates() usr_in = form['term_id'].value in_sample_type = form["sample_type"].value term_ids = get_searched_term_ids(usr_in) metasra_db = web.database(dbn='sqlite', db='static/metasra.sqlite') tsv_str = "sample_accession\tstudy_accession\tmapped_ontology_terms\tsample_type\tsample_type_confidence\n" n_results = 0 for term_id in term_ids: if in_sample_type == "all": results = query_metasra.query_metasra_for_term(metasra_db, term_id) else: results = query_metasra.query_metasra_for_term(metasra_db, term_id, sample_type=in_sample_type) for r in results: n_results += 1 sample_acc = r["sample_accession"] study_acc = r["study_accession"] r_sample_type = r["sample_type"] r_terms_csv = r["sample_terms_csv"] confidence = r["confidence"] tsv_str += "%s\t%s\t%s\t%s\t%0.3f\n" % (sample_acc, study_acc, r_terms_csv, r_sample_type, confidence) tsv_str = tsv_str[:-1] # remove trailing line-break print "Returning %d results" % n_results return tsv_str
def init_web_db(cfg_file = "config.cfg"): parser = ConfigParser.SafeConfigParser() parser.optionxform = str cn_db_file = os.getenv("CN_DB") if cn_db_file is not None: print "*** USING DATABASE", cn_db_file cfg_file = cn_db_file parser.read(cfg_file) section = 'database' if 'database' not in parser.sections(): raise Exception("No database section in " + cfg_file) dbn = parser.get(section, 'dbn') if dbn == "mysql": db = parser.get(section, 'db') user = parser.get(section, 'user') pw = parser.get(section, 'pw') host = parser.get(section, 'host') db = web.database(dbn='mysql', db=db, user=user, pw=pw, host=host) db.query('SET NAMES utf8;') db.query('SET CHARACTER SET utf8;') db.query('SET character_set_connection=utf8;') elif dbn == "sqlite": dbname = parser.get(section, 'db') db = web.database(dbn='sqlite', db=dbname) # We need to mimic some MySQL functions in order to be able to use # SQLite or use different SQL commands for each database server. I # prefer the 1st option, naturally... db._db_cursor().connection.create_function("concat", 2, sqlite_concat) db._db_cursor().connection.create_function("conv", 3, sqlite_conv) db._db_cursor().connection.create_function("instr", 2, sqlite_instr) db._db_cursor().connection.create_function("rand", 0, sqlite_rand) return db #-----------------------------------------------------------------------
def init_web_db(cfg_file="config.cfg"): parser = ConfigParser.SafeConfigParser() parser.optionxform = str parser.read(cfg_file) section = 'database' if 'database' not in parser.sections(): raise Exception("No database section in %s" % cfg_file) dbn = parser.get(section, 'dbn') if dbn == "mysql": db = parser.get(section, 'db') user = parser.get(section, 'user') pw = parser.get(section, 'pw') host = parser.get(section, 'host') db = web.database(dbn='mysql', db=db, user=user, pw=pw, host=host) db.query('SET NAMES utf8;') db.query('SET CHARACTER SET utf8;') db.query('SET character_set_connection=utf8;') elif dbn == "sqlite": dbname = parser.get(section, 'db') db = web.database(dbn='sqlite', db=dbname) # We need to mimic some MySQL functions in order to be able to use # SQLite or use different SQL commands for each database server. I # prefer the 1st option, naturally... db._db_cursor().connection.create_function("concat", 2, sqlite_concat) db._db_cursor().connection.create_function("conv", 3, sqlite_conv) db._db_cursor().connection.create_function("instr", 2, sqlite_instr) db._db_cursor().connection.create_function("rand", 0, sqlite_rand) return db #-------------------------------------------------------------------------------
def POST(self): form = search_form() form.validates() error_message = "" usr_in = form['term_id'].value metasra_db = web.database(dbn='sqlite', db='static/metasra.sqlite') #term_ancestors_db = web.database(dbn='sqlite', db='static/term_ancestors.sqlite') term_ids = get_searched_term_ids(usr_in) if len(term_ids) == 0: error_message = "The query did not match the name nor synonym of any ontology terms. Please use the search-suggestion feature to guide your search." request_results = [] for term_id in term_ids: in_sample_type = form["sample_type"].value if in_sample_type == "all": results = query_metasra.query_metasra_for_term(metasra_db, term_id) else: results = query_metasra.query_metasra_for_term(metasra_db, term_id, sample_type=in_sample_type) print "Preparing results..." for r in results: r_sample_acc = r["sample_accession"] r_study_acc = r["study_accession"] if r["study_title"]: r_study_title = r["study_title"] else: r_study_title = "" r_attrs_elem = r["sample_attributes_html"] r_terms_elem = r["sample_terms_html"] r_sample_name = r["sample_name"] r_sample_type = r["sample_type"] r_confidence = r["confidence"] request_results.append([ '<a class="sample_accession_link" target="_blank" href="https://www.ncbi.nlm.nih.gov/biosample/%s">%s</a>' % (r_sample_acc,r_sample_acc), '<a class="sample_accession_link" target="_blank" href="https://www.ncbi.nlm.nih.gov/biosample/%s">%s</a>' % (r_sample_acc,r_sample_name), '<a class="study_link" target="_blank" href="https://trace.ncbi.nlm.nih.gov/Traces/sra/?study=%s">%s</a>' % (r_study_acc, r_study_title), '<div class="sample_type_table_entry">%s (%0.2f)</div>' % (r_sample_type, r_confidence), r_terms_elem, r_attrs_elem ]) print "Finished preparing results." print "Returning %d results" % len(request_results) return json.dumps({"error_message":error_message, "search_results":request_results})
