SQLAlchemy Insert()从多个数据源

Question

我有两个表，提供有关员工和部门的信息。

我将从这些表中查询信息，并将其插入到新的第三个表中，该表必须包含其他信息，而不是预览中提供的信息。

stmt = select (
employees.columns['emp_id'],
employees.columns['f_name'],
departments.columns['dept_id_dep'],
departments.columns['dep_name']
)
.select_from(
employees.join(
departments,
employees.columns['dep_id'] == departments.columns['dept_id_dep'],
isouter=True
)
)
EandP = Table('EmployeesPlusDepart', metadata,
Column('Emp_id', String(50), primary_key = True, autoincrement = False), 
Column('Name', String (50), index = False, nullable = False), 
Column('Dept_id', String (50), nullable = False),
Column('Dept_Name', String (50), nullable = False),
Column('Location', String(50), default = 'CasaDuCarai', nullable = False), 
Column('Start_date', Date,
default = date.today() - timedelta(days=5), onupdate = date.today()),
extend_existing=True, #força a redefinição no metadata
)
Insert_stmt = insert(EandP).from_select(
['Emp_id', 'Name', 'Dept_id', 'Dept_Name'],
stmt
)

1. 新表列的位置和start_date我想提供手动在执行上面插入代码。那么，我应该如何将这些值聚合到上面的insert().from_select()中呢?

2. 查看select语句，我从同一个表中选择了几个列(例如:员工。列[' emp_id '], employees.columns [' f_name '])。我可以使用">employees.columns['emp_id'，'f_name']"让声明变小?

obs:我已经为Location和StartDate表设置了一个默认值，只是为了避免让它们为空

Answer 1

您可以通过创建带有自定义编译器行为的自定义表达式来实现这一点。在您的示例中，我假设您希望手动确定特定列的值—您可以在select语句中执行此操作:

from sqlalchemy.sql.expression import Executable, ClauseElement, literal

# Modify your select query to match the destination table
select_statement = select (
employees.columns['emp_id'].label('Emp_id'),
employees.columns['f_name'].label('Name'),
departments.columns['dept_id_dep'].label('Dept_id'),
departments.columns['dep_name'].label('Dept_id'),
# Note: I am not 100% sure about the expression below ("literal")
literal("Some_Location").label("Location")  # Manually set a location
)
.select_from(
employees.join(
departments,
employees.columns['dep_id'] == departments.columns['dept_id_dep'],
isouter=True
)
)

class InsertFromSelect(Executable, ClauseElement):
"""
Copied from SQLAlchemy Documentation:
https://docs.sqlalchemy.org/en/14/core/compiler.html#compiling-sub-elements-of-a-custom-expression-construct
"""
inherit_cache = False
def __init__(self, table, select):
self.table = table
self.select = select
@compiles(InsertFromSelect)
def _insert_from_select(element, compiler, **kwargs):
table_formatted = compiler.process(element.table, asfrom=True, **kwargs)
select_formatted = compiler.process(element.select, asfrom=True, **kwargs)
return f"INSERT INTO {table_formatted} ({select_formatted})"
# Create your insert statement
insert_statement = InsertFromSelect(
table=EandP,
select=select_statement
)
# Execute
session.execute(insert_statement)