SQLAlchemy Insert() from multiples sources

Question

I have 2 tables that provide information about employees and departments.

I will query information from those tables and insert it to a new third table, that must contains other info than such avaible from the previews 2.

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. new table as columns as location and start_date that I would like to provide manually when executing the insert code above. So, what should I aggregate such values to the insert().from_select() above?

2. Looking at the select statement, I choose several columns from the same table (e.g. employees.columns['emp_id'], employees.columns['f_name']). Can´t I use something like "employees.columns['emp_id','f_name']" the make the statement smaller?

obs : I´ve set up an default value for the Location and StartDate tables just to avoid leaving them null

Answer 1

You can accomplish this by creating a customized expression with customized compiler behavior. In your case, I'm assuming you want to manually determine the values of specific columns – you can do that in your select statement:

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)