SQLAlchemy 连接 2 个表

Question

我想创建 python SQLAlchemy，这是我数据库上的 2 个表，

table name : student

- id (integer)

-  student_name (varchar(20))

- edu_level_id (integer)

包含记录

id |  student_name  | edu_level_id

1  |  John  | 2

2  |  George | 1

表：master_edu_level

* id (integer)
* name_level (varchar(50))

记录：

id  |  name_level  |   

1   |  high school

2   |  Bachelor

3 |  Master

如何显示记录：

id |  student_name   | education


1  | John     | Bachelor

2  |  George  | high school

如果我这样解释 SQL：

select student.id, student.name,master_edu_level.name AS educatio FROM student
left join master_edu_Level ON master_edu_level.id = student.edu_level_id

我已经写好了代码：

from app import db
from app import session
from sqlalchemy import Table, DateTime,Date, Text, text, Column, Integer, String, MetaData, ForeignKey, select, func
from sqlalchemy.ext.declarative import declarative_base
from marshmallow_jsonapi import Schema, fields
from marshmallow import validate
from marshmallow import ValidationError
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import relationship

class Student(db.Model):
        __tablename__ = 'student'

        id = db.Column(Integer, primary_key=True, index=True)
        student_name = db.Column(String(20), nullable=False, index=True)
        edu_level_id = db.Column(Integer)

class StudentSchema(Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer(dump_only=True)
    student_name        = fields.String()  
    edu_level_id        = fields.String()  

    class Meta:
        type_ = 'student'

student_schema = StudentSchema()  


class MasterEduLevel(db.Model):
    __tablename__ = 'master_edu_level'
    __table_args__ = {'extend_existing': True} 

    id = db.Column(Integer, primary_key=True, index=True)
    name_level = db.Column(String(20), nullable=False, index=True)


class MasterEdulevelSchema(Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer(dump_only=True)
    name_level        = fields.String()  

    class Meta:
        type_ = 'master_edu_level'

schema = MasterEdulevelSchema()

如何在 python SQLAlchemy 上创建连接表 ORM？

谢谢

Answer 1

在第一步更改架构时：

import marshmallow as mm
from marshmallow_sqlalchemy import ModelSchema

class StudentSchema(ModelSchema):
    id = mm.fields.Integer(dump_only=True)
    student_name = mm.fields.String()
    education = mm.fields.Nested('MasterEdulevelSchema')

    class Meta:
        type_ = 'student'

    @mm.post_dump
    def format_dump(self, data):
        education = data.pop('education')
        data['education'] = education['name_level']
        return data


class MasterEdulevelSchema(ModelSchema):
    name_level = mm.fields.String()  

    class Meta:
        type_ = 'master_edu_level'

它将以正确的方式格式化您的数据。

下一步转储数据库实体：

student_schema = StudentSchema() 
student = db.session.query(Student).join(MasterEduLevel, Student.edu_level_id == MasterEduLevel.id).first()
student_result = student_schema.dump(student)

这应该可以，但我还没有测试过。

顺便说一句，有很多方法可以解决这个问题。与SQLAlchemy 一样，在Marshmallow 一侧。

Answer 2

所以，我决定使用文本查询 SQLalchemy 像这样

params = {"id":1}
s = text("SELECT student.*, master_edu_level.name_level FROM student left join master_edu_level ON master_edu_level.id = student.edu_level_id WHERE student.id= :id")
users_query= session.execute(s, params ).fetchall()
results = schema.dump(users_query, many=True).data

然后对于模式学生，添加字段 name_level

喜欢这个

class StudentSchema(Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer(dump_only=True)
    student_name        = fields.String()  
    edu_level_id        = fields.String()  
    name_level = fields.String()

谢谢。

Answer 3

试试这个：

from app import db
from app import session
from sqlalchemy import Table, DateTime,Date, Text, text, Column, Integer, String, MetaData, ForeignKey, select, func
from sqlalchemy.ext.declarative import declarative_base
from marshmallow_jsonapi import Schema, fields
from marshmallow import validate
from marshmallow import ValidationError
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import relationship


class MasterEduLevel(db.Model):
    __tablename__ = 'master_edu_level'
    __table_args__ = {'extend_existing': True} 

    id = db.Column(Integer, primary_key=True, index=True)
    name_level = db.Column(String(20), nullable=False, index=True)


class MasterEdulevelSchema(Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer(dump_only=True)
    name_level        = fields.String()  

    class Meta:
        type_ = 'master_edu_level'

class Student(db.Model):
    __tablename__ = 'student'

    id = db.Column(Integer, primary_key=True, index=True)
    student_name = db.Column(String(20), nullable=False, index=True)
    edu_level_id = db.Column(Integer, db.ForeignKey(MasterEduLevel.id))
    edu_level = db.relationship("MasterEduLevel")

class StudentSchema(Schema):
    not_blank = validate.Length(min=1, error='Field cannot be blank')
    id = fields.Integer(dump_only=True)
    student_name        = fields.String()  
    edu_level_id        = fields.String()

    edu_level           = fields.Function(lambda x: x.edu_level)  

    # You can use Nested for get the entire object of the MaterEduLevel (comment above edu_level)
    #edu_level           = fields.Nested("MasterEdulevelSchema")

    # You can also specify a field of the Nested object using the 'only' attribute (comment above edu_level)
    #edu_level           = fields.Nested("MasterEdulevelSchema", only=('name_level',))  

    class Meta:
        type_ = 'student'

student_schema = StudentSchema()
# You can also use the ONLY attribute here
#student_schema = StudentSchema(only=('student_name', 'edu_level'))

# Many results
users_query = session.query(Student).all()
results = student_schema.dump(users_query, many=True).data

# Many results with filter
users_query = session.query(Student).filter(Student.name.like('jo%')).all()
results = student_schema.dump(users_query, many=True).data

# Many results with filter in child property
users_query = session.query(Student).join(Student.edu_level).filter(MasterEduLevel.name == 'Bachelor').all()
results = student_schema.dump(users_query, many=True).data

# Just one row by PK
user = session.query(Student).get(1) #Student.id == 1
result = student_schema.dump(user).data