Context:
I'm working on a tool that allows my team to stream data for testing from a number of different sources to several different sink (e.g. s3, db, file system, spark, kafka, etc.). In the process, I need to also do some transformation of data for testing (e.g. pseudonymization and/or anonymization of personal data). In a non-testing scenario, I figure I should probably also have the option of just a straight copy/sync.

Initial Approach:
I had started with sqlx, and I had a solution that allowed me to stream from a source database, but I quickly ran into ergonomic issues trying to create generic push methods. It looks to me like sea-orm should be able to solve the ergonomic solutions. With raw sqlx, I had created a (set of) tokio channel(s) to do the following operations: pull, transform, capture metadata/metrics, push, validate. To allow for parallelization, I created an enumeration (Record) which had an entry for each of the types of tables rows I wanted to process. This worked really well for a pull. This took the form (more or less) of:

enum Record {
   FooTable(FooRecord), 
   BarTable(BarRecord),
   ...
   }

#[derive(Clone, Debug, Default, Deserialize, Serialize, FromRow)
struct FooRecord {
   baz: ...
   }
   
impl From<FooRecord> for Record {...}
   
async fn pull_tables(configuration: ..., tx_channel: ...) -> Result<()> {
    for table_name in configuration.table_names {
        match table_name {
            "foo" => pull_table_data<FooRecord>(...).await?;
        }
    }
}

async fn pull_table_data<T>(table_name: &str, tx_channel: ...) -> Result<()>
where 
    T: MyCustomDeriveForColumnIteratorAndTableName,
    Record: From<T> {
    ...
    let sql = "SELECT * FROM {table_name}";
    let stream = sqlx::query(sql).fetch(pool).await?;
    while let Some(stream) = stream.try_next().await {
        ...
        let raw_record: T = stream;
        let record = Record::from(raw_record);
        if let Err(why) = tx.send(record).await? {...}
    }
}

Problem:
I realized that I was going to need to use sqlx's .bind(...) a little late for a push. This causes a bunch of problems for a generic push and would really expand my derive macro. Rather than pushing through, I thought maybe I could find another solution. I stumbled on sea-orm and the interface seemed pretty close to what I wanted for the insert. So I replaced my model above with the expected sea-orm Model version and a DeriveEntityModel. This caused my where clause to require a change, and it ended up here:

#[derive(Clone, Debug, Default, Deserialize, Serialize, DeriveEntityModel)
struct Model {
   baz: ...
   }
   
impl From<foo::Model> for Record {...}

async fn pull_tables(configuration: ..., tx_channel: ...) -> Result<()> {
    ...
    "foo" => pull_table_data<foo::Model>(...).await?;
    ...
}

pub async fn pull_table_data<M>(table_names: &[String], dumper_config: &DbDumperConfig, storage_config: &StorageConfig, pool: &Pools, tx: Sender<MyRecord>) -> anyhow::Result<()>
where
    M: 'static + std::fmt::Debug + Default + Send + Unpin,
    MyRecord: From<M>
   {
       ...
       let stream = M::find().stream(pool).await?;
       ...
   }

This was fine, except when I called pull_table_data, it gave me an error: EntityTrait isn't implemented. Okay, easy... just add EntityTrait as a restriction.

    M: 'static + std::fmt::Debug + Default + Send + Unpin + EntityTrait,

Unfortunately, now there's a really strange (to me) error:

the trait `std::convert::From<<M as sea_orm::EntityTrait>::Model>` is not implemented for `models::record::MyRecord`

I'm not sure why there's a problem. Any chance someone can help me understand why I need to implement EntityTrait for MyRecord? It seems like they really should not be related.