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
python类database()的实例源码
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 __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 __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 __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 __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 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)
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})
