Skip to main content
 首页 » 编程设计

python之如何在 Python 中通过 API 查询在 SQL Alchemy 中插入关系数据(多对多)

2025年05月04日12yxwkf

编辑:我做了一个 short question因为我觉得这个太长了,抱歉

首先,我是数据库、编程语言等方面的新手...很抱歉,如果这个问题不太恰当或不具体,我们将不胜感激任何帮助或指导...

我正在使用的上下文如下:我正在通过其 API 查询现有数据库,以便检索某些信息来设计我自己的数据库。

创建此数据库的要点是,例如,让用户引入一个基因,以了解它在生物体中的哪个位置表达过度 (UP) 或表达不足 (DOWN),以及在哪个实验中看到了这种类型的表达。

目前,我正在做的只是查询现有数据库并解析 json 结果以获得每个生物体部分,所有过度表达或表达不足的基因(以及我获得的每个基因已报告该类型表达的实验)

(在大脑中)

基因1

Experiment1     UP 
Experiment2     UP 
Experiment3     UP 
Experiment4     DOWN 

基因2

Experiment5     DOWN 
Experiment2     DOWN 
Experiment3     DOWN 
Experiment8     UP 
Experiment9     DOWN 

我认为我需要的不同表格是:“基因”、“器官”、“实验”和“表达类型”(以及“基因2实验2器官”)

考虑到一个基因可以在多个生物体部分表达,并且可以具有与多个实验相关的不同类型的表达,并且一个实验可以包含多个基因(多对多关系)

我首先想知道的是如何添加关系数据,并知道我的尝试是否朝着正确的方向发展,或者我是否应该更改数据库的模式/想法...

我的第一次尝试是这样的:

########################################### 
DATABASE DEFINITION 
########################################### 
 
from sqlalchemy import create_engine, Column, Integer, String, Date, ForeignKey, Table, Float 
from sqlalchemy.orm import sessionmaker, relationship, backref 
from sqlalchemy.ext.declarative import declarative_base 
import requests 
 
Base = declarative_base() 
 
Genes2experiments2organs = Table('genes2experiments2organs',Base.metadata,                                                                                                                                                                                                      
  Column('gene_id', String, ForeignKey('genes.id')),                                                                                                                                                                                                        
  Column('experiment_id', String, ForeignKey('experiments.id')), 
  Column('organ_id', String, ForeignKey('organs.id')) 
) 
 
class Genes(Base): 
    __tablename__ = 'genes' 
    id = Column(String(45), primary_key=True) 
    def __init__(self, id=""): 
        self.id= id 
    def __repr__(self): 
        return "<genes(id:'%s')>" % (self.id) 
 
class Experiments(Base): 
    __tablename__ = 'experiments' 
    id = Column(String(45), primary_key=True) 
    experiments = relationship("Experiments", secondary=Genes2experiments2organs, backref="genes") 
    organs = relationship("Organs", secondary=Genes2experiments2organs, backref="genes") 
    def __init__(self, id=""): 
        self.id= id 
    def __repr__(self): 
        return "<experiments(id:'%s')>" % (self.id) 
 
class Organs(Base): 
    __tablename__ = 'organs' 
    id = Column(String(45), primary_key=True) 
    def __init__(self, id=""): 
        self.id= id 
    def __repr__(self): 
        return "<organs(id:'%s')>" % (self.id) 
 
class Expression_type(Base): 
    __tablename__ = 'expression_type' 
    id = Column(String(45), primary_key=True) 
    def __init__(self, id=""): 
        self.id= id 
    def __repr__(self): 
        return "<expression_type(id:'%s')>" % (self.id) 
 
##################################################### 
INSERTING DATA 
##################################################### 
 
def setUp(): 
    global Session 
    engine=create_engine('mysql://root:password@localhost/db_name?charset=utf8', pool_recycle=3600,echo=False) 
    Session=sessionmaker(bind=engine) 
 
def add_data():   ## I am just adding genes without taking into account the other related data to these genes..... 
    session=Session() 
    for i in range(0,1000,200): 
        request= requests.get('http://www.ebi.ac.uk/gxa/api/v1',params={"updownInOrganism_part":"brain","rows":200,"start":i}) 
        result = request.json 
        for item in result['results']: 
            gene_to_add = item['gene']['ensemblGeneId']            
    session.commit() 
    session.close()        
 
 
setUp() 
add_data() 
session=Session() 
genes=session.query(Genes).all() 
print "List of genes introduced:" 
for gene in genes: 
    print gene.id 
session.close() 

因此,使用此代码我只是填充“基因”表,但没有考虑与我必须包含在数据库中的其他数据存在的关系......做这样的事情的过程是什么,添加关系数据?还有一种避免插入重复基因的方法,例如在通过 API 查询填充表时??

顺便说一句,如您所见,我没有将所有多对多关系(次要关系)放在“基因”表中,因为我不确定我的做法是正确的还是完全错误的...谢谢

请您参考如下方法:

这应该可以满足您的需求...

from sqlalchemy import (Column, create_engine, Integer, ForeignKey, Unicode, 
                        Enum) 
from sqlalchemy.ext.declarative import declarative_base 
from sqlalchemy.orm import sessionmaker, relationship 
 
Base = declarative_base() 
 
class Gene(Base): 
    __tablename__ = 'gene' 
 
    id = Column(Integer, primary_key=True) 
    name = Column(Unicode(64), unique=True) 
 
    def __init__(self, name): 
        self.name = name 
 
class Experiment(Base): 
    __tablename__ = 'experiment' 
 
    id = Column(Integer, primary_key=True) 
 
class Organ(Base): 
    __tablename__ = 'organ' 
 
    id = Column(Integer, primary_key=True) 
    name = Column(Unicode(64), unique=True) 
 
    def __init__(self, name): 
        self.name = name 
 
class Measurement(Base): 
    __tablename__ = 'measurement' 
 
    id = Column(Integer, primary_key=True) 
    experiment_id = Column(Integer, ForeignKey(Experiment.id)) 
    gene_id = Column(Integer, ForeignKey(Gene.id)) 
    organ_id = Column(Integer, ForeignKey(Organ.id)) 
 
    # Add your measured values here 
    expression = Column(Enum('UP', 'DOWN')) 
    # ... 
 
    experiment = relationship(Experiment, backref='measurements') 
    gene = relationship(Gene, backref='measurements') 
    organ = relationship(Organ, backref='measurements') 
 
    def __repr__(self): 
        return 'Experiment %d: %s, %s, %s' % (self.experiment.id, 
                         self.gene.name, self.organ.name, self.expression) 
 
if __name__ == '__main__': 
    engine = create_engine('sqlite://') 
    session = sessionmaker(engine)() 
    Base.metadata.create_all(engine) 
 
    # 
    # Creating the data 
    # 
 
    x = Gene('Gene X') 
    y = Gene('Gene Y') 
    z = Gene('Gene Z') 
 
    heart = Organ('Heart') 
    lungs = Organ('Lungs') 
    brain = Organ('Brain') 
 
    session.add_all([x, y, z, heart, lungs, brain]) 
    session.commit() 
 
    experiment_1 = Experiment() 
    experiment_1.measurements.extend( 
            [Measurement(gene_id=x.id, organ_id=heart.id, expression='UP'), 
             Measurement(gene_id=x.id, organ_id=lungs.id, expression='UP'), 
             Measurement(gene_id=x.id, organ_id=brain.id, expression='DOWN'), 
             Measurement(gene_id=y.id, organ_id=brain.id, expression='UP'), 
             Measurement(gene_id=z.id, organ_id=brain.id, expression='DOWN')]) 
 
    experiment_2 = Experiment() 
    experiment_2.measurements.extend( 
            [Measurement(gene_id=y.id, organ_id=lungs.id, expression='UP'), 
             Measurement(gene_id=y.id, organ_id=lungs.id, expression='UP'), 
             Measurement(gene_id=y.id, organ_id=brain.id, expression='UP'), 
             Measurement(gene_id=x.id, organ_id=brain.id, expression='UP'), 
             Measurement(gene_id=z.id, organ_id=heart.id, expression='UP')]) 
 
    session.add_all([experiment_1, experiment_2]) 
    session.commit() 
 
    # 
    # Querying the data 
    # 
 
    print('All measurements in the first experiment') 
    experiment = session.query(Experiment).filter(Experiment.id == 1).one() 
    for measurement in experiment.measurements: 
        print(measurement) 
    print('') 
 
    print('All measurements of Gene X') 
    gene_x = session.query(Gene).filter(Gene.name == 'Gene X').one() 
    for measurement in gene_x.measurements: 
        print(measurement) 
    print('') 
 
    print('All measurements of the brain') 
    the_brain = session.query(Organ).filter(Organ.name == 'Brain').one() 
    for measurement in the_brain.measurements: 
        print(measurement) 
    print('')