use futures_util::TryStreamExt;
use datasources::*;
use datasources_memory::MemoryDriver;
use std::any::Any;
use std::sync::{Arc, Mutex};
fn record(pairs: &[(&str, Value)]) -> Row {
pairs
.iter()
.map(|(key, value)| ((*key).to_string(), value.clone()))
.collect()
}
async fn setup(rows: &mut [Row]) -> Arc<dyn Connection> {
let connection = MemoryDriver.connection(None);
connection.open().await.unwrap();
let command = connection.create_command();
for row in rows.iter_mut() {
let params: Vec<Parameter> = row
.iter()
.map(|(name, value)| Parameter::new(name.clone(), value.clone()))
.collect();
let response = command.execute("insert", Some(¶ms)).await.unwrap();
row.insert("id".into(), response.result);
}
connection
}
fn mapped_crud(connection: Arc<dyn Connection>, entity: &str) -> CRUDRepository {
mapped_crud_schema(connection, entity, Schema::default())
}
fn mapped_crud_schema(
connection: Arc<dyn Connection>,
entity: &str,
schema: Schema,
) -> CRUDRepository {
let mut options = CRUDRepositoryOptions::default();
options.entity = entity.into();
options.ids = vec!["Id".into()];
options.property_mappings = Some(std::sync::Arc::new(columns()));
options.schema = schema;
CRUDRepository::<Row>::new(connection, options).unwrap()
}
fn create_lookup_repo(connection: Arc<dyn Connection>, entity: &str) -> ReadonlyRepository {
let mut options = ReadonlyRepositoryMappingOptions::default();
options.entity = entity.into();
options.entity_mapper = Some(std::sync::Arc::new(ChangeCasingMapper::new(Casing::Pascal, Casing::Snake)));
options.property_mappings = Some(std::sync::Arc::new(ChangeCasingMapper::new(Casing::Pascal, Casing::Snake)));
ReadonlyRepository::<Row>::new(connection, options).unwrap()
}
fn columns() -> TransformMapper<String, String> {
TransformMapper::new(
|name: String| {
let references = DictionaryMapper::new([("Type".to_string(), "type_id".to_string())]);
let snake = ChangeCasingMapper::new(Casing::Pascal, Casing::Snake);
let mapped = references.map(name.clone());
if mapped != name {
mapped
} else {
snake.map(name)
}
},
|name: String| {
let references = DictionaryMapper::new([("Type".to_string(), "type_id".to_string())]);
let snake = ChangeCasingMapper::new(Casing::Pascal, Casing::Snake);
let from = references.reverse_map(name.clone());
if from != name {
from
} else {
snake.reverse_map(name)
}
},
)
}
fn mapped(connection: Arc<dyn Connection>, entity: &str) -> ReadonlyRepository {
let mut options = ReadonlyRepositoryMappingOptions::default();
options.entity = entity.into();
options.property_mappings = Some(std::sync::Arc::new(columns()));
ReadonlyRepository::<Row>::new(connection, options).unwrap()
}
fn num(row: &Row, key: &str) -> i64 {
match row.get(key) {
Some(Value::Int(id)) => *id,
_ => panic!("{key}"),
}
}
async fn drop_row(connection: &Arc<dyn Connection>, row: &Row) {
connection
.create_command()
.execute(
"delete",
Some(&[Parameter::new("id", row.get("id").unwrap().clone())]),
)
.await
.unwrap();
}
fn open_contacts(
mut contacts: CRUDRepository,
types: Arc<dyn IReadonlyRepository>,
sources: Arc<dyn IReadonlyRepository>,
address_types: Arc<dyn IReadonlyRepository>,
phone_types: Arc<dyn IReadonlyRepository>,
mut addresses: CRUDRepository,
mut phones: CRUDRepository,
) -> CRUDRepository {
addresses.add_enriched(
Enrichment::new("AddressType", "AddressType", "Id", None, datasources::EnrichmentPropertyType::Object),
address_types,
None,
);
phones.add_enriched(
Enrichment::new("PhoneType", "PhoneType", "Id", None, datasources::EnrichmentPropertyType::Object),
phone_types,
None,
);
contacts
.add_enriched(
Enrichment::new("ContactType", "ContactType", "Id", None, datasources::EnrichmentPropertyType::Object),
types,
None,
)
.add_enriched(
Enrichment::new("ContactSource", "ContactSource", "Id", None, datasources::EnrichmentPropertyType::Object),
sources,
None,
)
.add_enriched(
Enrichment::new("Addresses", "Id", "Contact", None, datasources::EnrichmentPropertyType::List),
Arc::new(addresses),
None,
)
.add_enriched(
Enrichment::new("Phones", "Id", "Contact", None, datasources::EnrichmentPropertyType::List),
Arc::new(phones),
None,
);
contacts
}
#[tokio::test]
async fn enriched_fills_lookup_select_and_leaves_missing_default() {
let mut seeds = [
record(&[("name", Value::from("person"))]),
record(&[("name", Value::from("gone"))]),
];
let types_connection = setup(&mut seeds).await;
let person = seeds[0].clone();
let gone = seeds[1].clone();
drop_row(&types_connection, &gone).await;
let mut parents = [
record(&[("type_id", person.get("id").unwrap().clone())]),
record(&[("type_id", gone.get("id").unwrap().clone())]),
];
let connection = setup(&mut parents).await;
let mut parents_repo = mapped(connection, "Parent");
let types_repo = Arc::new(mapped(types_connection, "UserType"));
parents_repo.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types_repo,
None,
);
let items = parents_repo.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(items[0]["TypeName"], Value::from("person"));
assert!(!items[1].contains_key("TypeName") || items[1]["TypeName"] == Value::from(""));
}
#[tokio::test]
async fn deep_create_writes_children_stamps_fk() {
let connection = setup(&mut []).await;
let mut parents = mapped_crud(connection.clone(), "Parent");
let types = Arc::new(mapped_crud(connection.clone(), "UserType"));
let children = Arc::new(mapped_crud(connection, "Child"));
let kind = types
.create(record(&[("Name", Value::from("person"))]))
.await
.unwrap();
assert!(num(&kind, "Id") > 0);
parents
.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types.clone(),
None,
)
.add_enriched(
Enrichment::new("Children", "Id", "ParentId", None, datasources::EnrichmentPropertyType::List),
children.clone(),
None,
);
let created = parents
.create(record(&[
("Type", kind.get("Id").unwrap().clone()),
("Label", Value::from("p")),
(
"Children",
encode_rows(&[
record(&[("Label", Value::from("a"))]),
record(&[("Label", Value::from("b"))]),
]),
),
]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
let stored_parents = parents.read(None, None).try_collect::<Vec<_>>().await.unwrap();
let stored = children.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(stored_parents[0]["Id"], created["Id"]);
assert_eq!(stored.len(), 2);
assert_eq!(stored[0]["ParentId"], stored_parents[0]["Id"]);
assert_eq!(stored[1]["ParentId"], stored_parents[0]["Id"]);
assert_eq!(
types.read(None, None).try_collect::<Vec<_>>().await.unwrap()[0]["Name"],
Value::from("person")
);
}
#[tokio::test]
async fn create_resolves_lookup_name() {
let texts = Arc::new(Mutex::new(Vec::new()));
let connection: Arc<dyn Connection> = Arc::new(RecordingConnection::new(
setup(&mut []).await,
Arc::clone(&texts),
));
let types = Arc::new(mapped_crud(Arc::clone(&connection), "UserType"));
let mut parents = mapped_crud(connection, "Parent");
types
.create(record(&[("Name", Value::from("work"))]))
.await
.unwrap();
parents.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types,
None,
);
let created = parents
.create(record(&[("TypeName", Value::from("work"))]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
assert_eq!(created["TypeName"], Value::from("work"));
assert_eq!(
parents.read_single(None, None).await.unwrap().unwrap()["TypeName"],
Value::from("work")
);
let sql = texts.lock().expect("texts");
let insert: Vec<_> = sql
.iter()
.filter(|text| text.starts_with("INSERT INTO Parent "))
.collect();
assert_eq!(insert.len(), 1);
assert_eq!(
column_list(insert[0])
.into_iter()
.filter(|name| *name == "type_id")
.count(),
1
);
}
#[tokio::test]
async fn update_resolves_lookup_name() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "UserType"));
let mut parents = mapped_crud(connection, "Parent");
let home = types
.create(record(&[("Name", Value::from("home"))]))
.await
.unwrap();
types
.create(record(&[("Name", Value::from("work"))]))
.await
.unwrap();
parents.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types,
None,
);
let created = parents
.create(record(&[("Type", home.get("Id").unwrap().clone())]))
.await
.unwrap();
parents
.update(record(&[
("Id", created.get("Id").unwrap().clone()),
("TypeName", Value::from("work")),
]), None)
.await
.unwrap();
assert_eq!(
parents.read_single(None, None).await.unwrap().unwrap()["TypeName"],
Value::from("work")
);
}
#[tokio::test]
async fn create_keeps_existing_lookup_key() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "UserType"));
let mut parents = mapped_crud(connection, "Parent");
let home = types
.create(record(&[("Name", Value::from("home"))]))
.await
.unwrap();
types
.create(record(&[("Name", Value::from("work"))]))
.await
.unwrap();
parents.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types.clone(),
None,
);
parents
.create(record(&[
("Type", home.get("Id").unwrap().clone()),
("TypeName", Value::from("work")),
]))
.await
.unwrap();
assert_eq!(
parents.read_single(None, None).await.unwrap().unwrap()["TypeName"],
Value::from("home")
);
assert_eq!(
types.read(None, None).try_collect::<Vec<_>>().await.unwrap().len(),
2
);
}
#[tokio::test]
async fn create_throws_when_two_keys_map_to_one_column() {
let connection = setup(&mut []).await;
let mut options = CRUDRepositoryOptions::default();
options.property_mappings = Some(Arc::new(DictionaryMapper::new([
("left".into(), "slot".into()),
("right".into(), "slot".into()),
])));
let repo = CRUDRepository::<Row>::new(connection, options).unwrap();
let error = repo
.create(record(&[
("id", Value::from(1)),
("left", Value::from(1)),
("right", Value::from(2)),
]))
.await
.unwrap_err();
assert!(error.to_string().contains("slot"));
}
#[tokio::test]
async fn create_resolves_lookup_name_without_source_property() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "UserType"));
let mut parents = mapped_crud(connection, "Named");
types
.create(record(&[("Name", Value::from("work"))]))
.await
.unwrap();
parents.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types,
None,
);
let created = parents
.create(record(&[("TypeName", Value::from("work"))]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
assert_eq!(created["TypeName"], Value::from("work"));
assert_eq!(
parents.read_single(None, None).await.unwrap().unwrap()["TypeName"],
Value::from("work")
);
}
#[tokio::test]
async fn create_unknown_lookup_name_throws() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "UserType"));
let mut parents = mapped_crud(connection, "Parent");
parents.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types,
None,
);
let error = parents
.create(record(&[("TypeName", Value::from("missing"))]))
.await
.unwrap_err();
assert!(error.to_string().contains("TypeName"));
}
#[tokio::test]
async fn create_missing_required_lookup_throws() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "UserType"));
let mut parents = mapped_crud_schema(
connection,
"Parent",
Schema::new(std::collections::HashMap::from([
(
"Id".into(),
SchemaColumn {
is_id: true,
..SchemaColumn::new("integer")
},
),
("TypeName".into(), SchemaColumn::new("string")),
])),
);
parents.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types,
None,
);
let error = parents.create(Row::new()).await.unwrap_err();
assert!(error.to_string().contains("TypeName"));
}
#[tokio::test]
async fn create_missing_nullable_lookup_skips() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "UserType"));
let mut parents = mapped_crud_schema(
connection,
"Parent",
Schema::new(std::collections::HashMap::from([
(
"Id".into(),
SchemaColumn {
is_id: true,
..SchemaColumn::new("integer")
},
),
(
"TypeName".into(),
SchemaColumn {
nullable: true,
..SchemaColumn::new("string")
},
),
])),
);
parents.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types,
None,
);
let created = parents.create(Row::new()).await.unwrap();
assert!(num(&created, "Id") > 0);
}
#[tokio::test]
async fn nested_child_resolves_lookup_name() {
let connection = setup(&mut []).await;
let mut parents = mapped_crud(connection.clone(), "Parent");
let mut children = mapped_crud(connection.clone(), "Child");
let types = Arc::new(mapped_crud(connection, "UserType"));
types
.create(record(&[("Name", Value::from("home"))]))
.await
.unwrap();
children.add_enriched(
Enrichment::new("TypeName", "Type", "Id", Some("Name".into()), datasources::EnrichmentPropertyType::Object),
types,
None,
);
let children = Arc::new(children);
parents.add_enriched(
Enrichment::new("Children", "Id", "ParentId", None, datasources::EnrichmentPropertyType::List),
children.clone(),
None,
);
parents
.create(record(&[(
"Children",
encode_rows(&[record(&[
("TypeName", Value::from("home")),
("Label", Value::from("a")),
])]),
)]))
.await
.unwrap();
let child = children
.read_single(None, None)
.await
.unwrap()
.unwrap();
assert_eq!(child["Label"], Value::from("a"));
assert_eq!(child["TypeName"], Value::from("home"));
}
#[tokio::test]
async fn round_trips_string_and_object_dictionaries() {
let connection = setup(&mut []).await;
let mut hosts = mapped_crud(connection.clone(), "Host");
let id = SchemaColumn {
is_id: true,
..SchemaColumn::new("integer")
};
let settings = Arc::new(mapped_crud_schema(
connection.clone(),
"Setting",
Schema::new(
[
("Id", id.clone()),
("ParentId", SchemaColumn::new("integer")),
("Name", SchemaColumn::new("string")),
("Value", SchemaColumn::new("string")),
]
.into_iter()
.map(|(name, column)| (name.to_string(), column))
.collect(),
),
));
let prefs = Arc::new(mapped_crud_schema(
connection,
"Pref",
Schema::new(
[
("Id", id),
("ParentId", SchemaColumn::new("integer")),
("Name", SchemaColumn::new("string")),
("Locale", SchemaColumn::new("string")),
("Timezone", SchemaColumn::new("string")),
]
.into_iter()
.map(|(name, column)| (name.to_string(), column))
.collect(),
),
));
hosts
.add_enriched(
Enrichment::new(
"Settings",
"Id",
"ParentId",
Some("Value".into()),
datasources::EnrichmentPropertyType::Dictionary,
)
.with_dictionary_key("Name"),
settings.clone(),
None,
)
.add_enriched(
Enrichment::new(
"Prefs",
"Id",
"ParentId",
None,
datasources::EnrichmentPropertyType::Dictionary,
)
.with_dictionary_key("Name"),
prefs.clone(),
None,
);
let created = hosts
.create(record(&[
("Label", Value::from("ada")),
(
"Settings",
encode_rows(&[
record(&[("Name", Value::from("theme")), ("Value", Value::from("dark"))]),
record(&[
("Name", Value::from("density")),
("Value", Value::from("comfortable")),
]),
]),
),
(
"Prefs",
encode_rows(&[
record(&[
("Name", Value::from("home")),
("Locale", Value::from("en-GB")),
("Timezone", Value::from("Europe/London")),
]),
record(&[
("Name", Value::from("travel")),
("Locale", Value::from("en-US")),
("Timezone", Value::from("America/New_York")),
]),
]),
),
]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
let loaded = hosts.read_single(None, None).await.unwrap().unwrap();
let settings_text = loaded["Settings"].wire_text();
assert!(settings_text.contains("Name=theme"));
assert!(settings_text.contains("Value=dark"));
assert!(settings_text.contains("Name=density"));
assert!(settings_text.contains("Value=comfortable"));
let prefs_text = loaded["Prefs"].wire_text();
assert!(prefs_text.contains("Name=home"));
assert!(prefs_text.contains("Timezone=Europe/London"));
assert!(prefs_text.contains("Name=travel"));
let updated_row_1 = record(&[
("Id", created["Id"].clone()),
("Label", Value::from("ada")),
(
"Settings",
encode_rows(&[
record(&[("Name", Value::from("theme")), ("Value", Value::from("light"))]),
record(&[("Name", Value::from("locale")), ("Value", Value::from("en"))]),
]),
),
(
"Prefs",
encode_rows(&[
record(&[
("Name", Value::from("home")),
("Locale", Value::from("en-GB")),
("Timezone", Value::from("Europe/Paris")),
]),
record(&[
("Name", Value::from("work")),
("Locale", Value::from("en-US")),
("Timezone", Value::from("America/New_York")),
]),
]),
),
]);
hosts
.update(updated_row_1, None)
.await
.unwrap();
let updated = hosts.read_single(None, None).await.unwrap().unwrap();
let settings_text = updated["Settings"].wire_text();
assert!(settings_text.contains("Name=theme"));
assert!(settings_text.contains("Value=light"));
assert!(settings_text.contains("Name=locale"));
assert!(!settings_text.contains("Name=density"));
let prefs_text = updated["Prefs"].wire_text();
assert!(prefs_text.contains("Timezone=Europe/Paris"));
assert!(prefs_text.contains("Name=work"));
assert!(!prefs_text.contains("Name=travel"));
let settings_rows = settings.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(settings_rows.len(), 2);
let pref_rows = prefs.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(pref_rows.len(), 2);
}
#[tokio::test]
async fn contacts_service_fills_names_and_child_lists() {
let mut person = [record(&[("name", Value::from("person"))])];
let types_connection = setup(&mut person).await;
let mut source = [record(&[("name", Value::from("import"))])];
let sources_connection = setup(&mut source).await;
let mut home = [record(&[("name", Value::from("Home"))])];
let address_types_connection = setup(&mut home).await;
let mut work = [record(&[("name", Value::from("Work"))])];
let phone_types_connection = setup(&mut work).await;
let connection = setup(&mut []).await;
let contacts = domain_crud(connection.clone(), "Contact");
let addresses = domain_crud(connection.clone(), "Address");
let phones = domain_crud(connection, "Phone");
let other = contacts
.create(record(&[
("ContactType", person[0].get("id").unwrap().clone()),
("ContactSource", source[0].get("id").unwrap().clone()),
("FirstName", Value::from("Other")),
]))
.await
.unwrap();
assert!(num(&other, "Id") > 0);
let stray = addresses
.create(record(&[
("Contact", other.get("Id").unwrap().clone()),
("Line1", Value::from("other")),
("City", Value::from("x")),
("AddressType", home[0].get("id").unwrap().clone()),
]))
.await
.unwrap();
assert!(num(&stray, "Id") > 0);
let service = open_contacts(
contacts,
Arc::new(create_lookup_repo(types_connection, "ContactType")),
Arc::new(create_lookup_repo(sources_connection, "ContactSource")),
Arc::new(create_lookup_repo(address_types_connection, "AddressType")),
Arc::new(create_lookup_repo(phone_types_connection, "PhoneType")),
addresses,
phones,
);
let created = service
.create(record(&[
("ContactType", person[0].get("id").unwrap().clone()),
("ContactSource", source[0].get("id").unwrap().clone()),
("FirstName", Value::from("Ada")),
(
"Addresses",
encode_rows(&[record(&[
("Line1", Value::from("1 Main")),
("City", Value::from("Paris")),
("AddressType", home[0].get("id").unwrap().clone()),
])]),
),
(
"Phones",
encode_rows(&[record(&[
("Number", Value::from("555")),
("PhoneType", work[0].get("id").unwrap().clone()),
])]),
),
]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
let contact = service
.read(None, None)
.try_collect::<Vec<_>>().await
.unwrap()
.into_iter()
.find(|item| item.get("FirstName") == Some(&Value::from("Ada")))
.unwrap();
assert_eq!(contact["Id"], created["Id"]);
assert_ne!(contact["Id"], other["Id"]);
assert!(contact["ContactType"].wire_text().contains("Name=person")
|| contact["ContactType"].wire_text().contains("name=person"));
assert!(contact["ContactSource"].wire_text().contains("Name=import")
|| contact["ContactSource"].wire_text().contains("name=import"));
let addresses = contact["Addresses"].wire_text();
assert!(addresses.contains("Line1=1 Main"), "{addresses}");
assert!(
addresses.contains("Name=Home")
|| addresses.contains("name=Home")
|| addresses.contains("Name%3DHome"),
"{addresses}"
);
assert!(!contact["Addresses"].wire_text().contains("Line1=other"));
let phones = contact["Phones"].wire_text();
assert!(phones.contains("Number=555"), "{phones}");
assert!(
phones.contains("Name=Work") || phones.contains("name=Work") || phones.contains("Name%3DWork"),
"{phones}"
);
}
fn domain_crud(connection: Arc<dyn Connection>, entity: &str) -> CRUDRepository {
let mut options = CRUDRepositoryOptions::default();
options.entity = entity.into();
options.ids = vec!["Id".into()];
options.entity_mapper = Some(std::sync::Arc::new(ChangeCasingMapper::new(Casing::Pascal, Casing::Snake)));
options.property_mappings = Some(std::sync::Arc::new(contact_columns()));
options.object_converter = Some(Box::new(
MetaDataRowTypeConverter::new()
.with_references(contact_references(), Box::new(contact_columns())),
));
CRUDRepository::<Row>::new(connection, options).unwrap()
}
fn contact_references() -> Vec<String> {
["Contact", "AddressType", "ContactType", "ContactSource", "PhoneType"]
.into_iter()
.map(str::to_string)
.collect()
}
fn contact_columns() -> TransformMapper<String, String> {
let references = std::sync::Arc::new(DictionaryMapper::new([
("Contact".to_string(), "contact_id".to_string()),
("AddressType".to_string(), "address_type_id".to_string()),
("ContactType".to_string(), "contact_type_id".to_string()),
("ContactSource".to_string(), "contact_source_id".to_string()),
("PhoneType".to_string(), "phone_type_id".to_string()),
("Line1".to_string(), "line1".to_string()),
("Line2".to_string(), "line2".to_string()),
]));
let snake = std::sync::Arc::new(ChangeCasingMapper::new(Casing::Pascal, Casing::Snake));
let forward_references = references.clone();
let forward_snake = snake.clone();
let reverse_references = references.clone();
let reverse_snake = snake;
TransformMapper::new(
move |name: String| {
let mapped = Mapper::map(forward_references.as_ref(), name.clone());
if mapped != name {
mapped
} else {
Mapper::map(forward_snake.as_ref(), name)
}
},
move |name: String| {
let from = Mapper::reverse_map(reverse_references.as_ref(), name.clone());
if from != name {
from
} else {
Mapper::reverse_map(reverse_snake.as_ref(), name)
}
},
)
}
#[tokio::test]
async fn deep_update_replace_syncs_children() {
let connection = setup(&mut []).await;
let mut parents = mapped_crud(connection.clone(), "Parent");
let children = Arc::new(mapped_crud(connection, "Child"));
let parent = parents
.create(record(&[("Label", Value::from("p"))]))
.await
.unwrap();
assert!(num(&parent, "Id") > 0);
let parent_id = parent.get("Id").unwrap().clone();
let kept = children
.create(record(&[
("ParentId", parent_id.clone()),
("Label", Value::from("a")),
]))
.await
.unwrap();
assert!(num(&kept, "Id") > 0);
let dropped = children
.create(record(&[
("ParentId", parent_id.clone()),
("Label", Value::from("b")),
]))
.await
.unwrap();
assert!(num(&dropped, "Id") > 0);
parents.add_enriched(
Enrichment::new("Children", "Id", "ParentId", None, datasources::EnrichmentPropertyType::List),
children.clone(),
None,
);
let updated_row_2 = record(&[
("Id", parent_id.clone()),
("Label", Value::from("p")),
(
"Children",
encode_rows(&[
record(&[
("Id", kept.get("Id").unwrap().clone()),
("Label", Value::from("aa")),
]),
record(&[("Label", Value::from("c"))]),
]),
),
]);
let updated = parents
.update(updated_row_2, None)
.await
.unwrap();
assert_eq!(updated.rows_affected, 1);
let stored = children.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(stored.len(), 2);
assert_eq!(
stored
.iter()
.find(|row| row.get("Id") == kept.get("Id"))
.unwrap()["Label"],
Value::from("aa")
);
assert_eq!(
stored
.iter()
.find(|row| row.get("Id") != kept.get("Id"))
.unwrap()["Label"],
Value::from("c")
);
assert!(stored
.iter()
.all(|row| row.get("ParentId") == Some(&parent_id)));
}
#[tokio::test]
async fn cascade_update_deletes_children_and_recreates() {
let connection = setup(&mut []).await;
let mut parents = mapped_crud(connection.clone(), "Parent").with_write(Arc::new(
DefaultWriteEnrichmentStrategy::new(
None,
Some(Arc::new(CascadeDeleteUpdateEnrichmentStrategy)),
None,
false,
),
));
let mut children_repo = mapped_crud(connection.clone(), "Child");
let notes = Arc::new(mapped_crud(connection, "Note"));
let parent = parents
.create(record(&[("Label", Value::from("p"))]))
.await
.unwrap();
assert!(num(&parent, "Id") > 0);
let parent_id = parent.get("Id").unwrap().clone();
let child = children_repo
.create(record(&[
("ParentId", parent_id.clone()),
("Label", Value::from("a")),
]))
.await
.unwrap();
assert!(num(&child, "Id") > 0);
let other = children_repo
.create(record(&[("Label", Value::from("other"))]))
.await
.unwrap();
assert!(num(&other, "Id") > 0);
let child_id = child.get("Id").unwrap().clone();
let note = notes
.create(record(&[
("ChildId", child_id.clone()),
("Text", Value::from("n")),
]))
.await
.unwrap();
assert!(num(¬e, "Id") > 0);
children_repo.add_enriched(
Enrichment::new("Notes", "Id", "ChildId", None, datasources::EnrichmentPropertyType::List),
notes.clone(),
None,
);
let children = Arc::new(children_repo);
parents.add_enriched(
Enrichment::new("Children", "Id", "ParentId", None, datasources::EnrichmentPropertyType::List),
children.clone(),
None,
);
let updated_row_3 = record(&[
("Id", parent_id.clone()),
("Label", Value::from("p")),
(
"Children",
encode_rows(&[record(&[
("Id", child_id.clone()),
("Label", Value::from("fresh")),
])]),
),
]);
let updated = parents
.update(updated_row_3, None)
.await
.unwrap();
assert_eq!(updated.rows_affected, 1);
let stored = children.read(None, None).try_collect::<Vec<_>>().await.unwrap();
let created = stored
.iter()
.find(|row| row.get("Label") == Some(&Value::from("fresh")))
.unwrap();
assert_eq!(stored.len(), 2);
assert_eq!(created.get("Id"), Some(&child_id));
assert_eq!(created.get("Label"), Some(&Value::from("fresh")));
assert_eq!(created.get("ParentId"), Some(&parent_id));
assert!(notes.read(None, None).try_collect::<Vec<_>>().await.unwrap().is_empty());
}
#[tokio::test]
async fn cascade_update_replaces_links_and_keeps_related() {
let connection = setup(&mut []).await;
let mut parents = mapped_crud(connection.clone(), "Parent").with_write(Arc::new(
DefaultWriteEnrichmentStrategy::new(
None,
Some(Arc::new(CascadeDeleteUpdateEnrichmentStrategy)),
None,
false,
),
));
let tags = Arc::new(mapped_crud(connection.clone(), "Tag"));
let links = Arc::new(mapped_crud(connection, "ParentTag"));
let parent = parents
.create(record(&[("Label", Value::from("p"))]))
.await
.unwrap();
assert!(num(&parent, "Id") > 0);
let parent_id = parent.get("Id").unwrap().clone();
let keep = tags
.create(record(&[("Name", Value::from("keep"))]))
.await
.unwrap();
assert!(num(&keep, "Id") > 0);
let drop = tags
.create(record(&[("Name", Value::from("drop"))]))
.await
.unwrap();
assert!(num(&drop, "Id") > 0);
let keep_id = keep.get("Id").unwrap().clone();
let drop_id = drop.get("Id").unwrap().clone();
let old_link = links
.create(record(&[
("ParentId", parent_id.clone()),
("TagId", keep_id.clone()),
]))
.await
.unwrap();
assert!(num(&old_link, "Id") > 0);
let dropped_link = links
.create(record(&[
("ParentId", parent_id.clone()),
("TagId", drop_id),
]))
.await
.unwrap();
assert!(num(&dropped_link, "Id") > 0);
let leftover = links
.create(record(&[("TagId", keep_id.clone())]))
.await
.unwrap();
assert!(num(&leftover, "Id") > 0);
let old_link_id = old_link.get("Id").unwrap().clone();
parents.add_linked(
Enrichment::new("Tags", "Id", "Id", None, datasources::EnrichmentPropertyType::List)
.linked(Association::new("ParentId", "TagId")),
tags.clone(),
links.clone(),
None,
);
let updated_row_4 = record(&[
("Id", parent_id.clone()),
("Label", Value::from("p")),
("Tags", encode_rows(&[record(&[("Id", keep_id.clone())])])),
]);
let updated = parents
.update(updated_row_4, None)
.await
.unwrap();
assert_eq!(updated.rows_affected, 1);
let stored_links: Vec<_> = links
.read(None, None)
.try_collect::<Vec<_>>().await
.unwrap()
.into_iter()
.filter(|row| row.get("ParentId") == Some(&parent_id))
.collect();
assert_eq!(stored_links.len(), 1);
assert_eq!(stored_links[0].get("TagId"), Some(&keep_id));
assert_eq!(stored_links[0].get("ParentId"), Some(&parent_id));
assert_ne!(stored_links[0].get("Id"), Some(&old_link_id));
assert_eq!(tags.read(None, None).try_collect::<Vec<_>>().await.unwrap().len(), 2);
}
#[tokio::test]
async fn bulk_update_keeps_earlier_write_when_transaction_is_off() {
let connection = setup(&mut []).await;
let mut options = CRUDRepositoryOptions::default();
options.entity = "Parent".into();
let parents = BulkCRUDRepository::new(connection, options).unwrap();
let created = parents
.create_bulk(&mut [
record(&[("label", Value::from("a"))]),
record(&[("label", Value::from("b"))]),
])
.await
.unwrap();
let stored = parents.read(None, None).try_collect::<Vec<_>>().await.unwrap();
let mut created_ids: Vec<_> = created.iter().map(|row| num(row, "id")).collect();
let mut stored_ids: Vec<_> = stored.iter().map(|row| num(row, "id")).collect();
created_ids.sort();
stored_ids.sort();
assert_eq!(created_ids, stored_ids);
let first_id = stored
.iter()
.find(|row| row.get("label") == Some(&Value::from("a")))
.unwrap()
.get("id")
.unwrap()
.clone();
let error = parents
.update_bulk(
&[
record(&[("id", first_id.clone()), ("label", Value::from("changed"))]),
record(&[("label", Value::from("nope"))]),
],
None,
)
.await
.unwrap_err();
assert!(error.to_string().contains("Missing Id"));
let after = parents.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(
after
.iter()
.find(|row| row.get("id") == Some(&first_id))
.unwrap()
.get("label"),
Some(&Value::from("changed"))
);
}
#[tokio::test]
async fn bulk_update_rolls_back_when_transaction_is_on() {
let connection = setup(&mut []).await;
let mut options = CRUDRepositoryOptions::default();
options.entity = "Parent".into();
options.write = Some(Arc::new(DefaultWriteEnrichmentStrategy::new(
None, None, None, true,
)));
let parents = BulkCRUDRepository::new(connection, options).unwrap();
let created = parents
.create_bulk(&mut [
record(&[("label", Value::from("a"))]),
record(&[("label", Value::from("b"))]),
])
.await
.unwrap();
let stored = parents.read(None, None).try_collect::<Vec<_>>().await.unwrap();
let mut created_ids: Vec<_> = created.iter().map(|row| num(row, "id")).collect();
let mut stored_ids: Vec<_> = stored.iter().map(|row| num(row, "id")).collect();
created_ids.sort();
stored_ids.sort();
assert_eq!(created_ids, stored_ids);
let first_id = stored
.iter()
.find(|row| row.get("label") == Some(&Value::from("a")))
.unwrap()
.get("id")
.unwrap()
.clone();
let error = parents
.update_bulk(
&[
record(&[("id", first_id.clone()), ("label", Value::from("changed"))]),
record(&[("label", Value::from("nope"))]),
],
None,
)
.await
.unwrap_err();
assert!(error.to_string().contains("Missing Id"));
let after = parents.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(
after
.iter()
.find(|row| row.get("id") == Some(&first_id))
.unwrap()
.get("label"),
Some(&Value::from("a"))
);
}
#[tokio::test]
async fn deep_delete_cascades_children_then_parent() {
let connection = setup(&mut []).await;
let parents = mapped_crud(connection.clone(), "Parent");
let mut children_repo = mapped_crud(connection.clone(), "Child");
let notes = Arc::new(mapped_crud(connection, "Note"));
let parent = parents
.create(record(&[("Label", Value::from("p"))]))
.await
.unwrap();
assert!(num(&parent, "Id") > 0);
let parent_id = parent.get("Id").unwrap().clone();
let child = children_repo
.create(record(&[
("ParentId", parent_id.clone()),
("Label", Value::from("a")),
]))
.await
.unwrap();
assert!(num(&child, "Id") > 0);
let child_id = child.get("Id").unwrap().clone();
let note = notes
.create(record(&[
("ChildId", child_id),
("Text", Value::from("n")),
]))
.await
.unwrap();
assert!(num(¬e, "Id") > 0);
let other = notes
.create(record(&[
("Text", Value::from("other")),
("Label", Value::from("keep")),
]))
.await
.unwrap();
assert!(num(&other, "Id") > 0);
children_repo.add_enriched(
Enrichment::new("Notes", "Id", "ChildId", None, datasources::EnrichmentPropertyType::List),
notes.clone(),
None,
);
let children = Arc::new(children_repo);
let mut parents = parents;
parents.add_enriched(
Enrichment::new("Children", "Id", "ParentId", None, datasources::EnrichmentPropertyType::List),
children.clone(),
None,
);
let removed = parents
.delete(record(&[("Id", parent_id)]), None)
.await
.unwrap();
assert_eq!(removed.rows_affected, 1);
assert!(parents.read(None, None).try_collect::<Vec<_>>().await.unwrap().is_empty());
assert!(children.read(None, None).try_collect::<Vec<_>>().await.unwrap().is_empty());
assert_eq!(
notes.read(None, None).try_collect::<Vec<_>>().await.unwrap()[0]["Text"],
Value::from("other")
);
}
#[tokio::test]
async fn deep_many_readonly_via_throws() {
let connection = setup(&mut []).await;
let mut parents = mapped_crud(connection.clone(), "Parent");
parents.add_enriched(
Enrichment::new("Children", "Id", "ParentId", None, datasources::EnrichmentPropertyType::List),
Arc::new(mapped(connection, "Child")),
None,
);
let error = parents
.create(record(&[(
"Children",
encode_rows(&[record(&[("Label", Value::from("a"))])]),
)]))
.await
.unwrap_err();
assert!(error.to_string().contains("Children"));
}
#[tokio::test]
async fn bulk_create_writes_parents_and_children() {
let connection = setup(&mut []).await;
let mut options = CRUDRepositoryOptions::default();
options.entity = "Parent".into();
let mut parents = BulkCRUDRepository::new(connection.clone(), options).unwrap();
let mut child_options = CRUDRepositoryOptions::default();
child_options.entity = "Child".into();
let children = Arc::new(BulkCRUDRepository::new(connection, child_options).unwrap());
parents.add_enriched(
Enrichment::new("children", "id", "parent_id", None, datasources::EnrichmentPropertyType::List),
children.clone(),
None,
);
let ids = parents
.create_bulk(&mut [
record(&[(
"children",
encode_rows(&[record(&[("label", Value::from("a"))])]),
)]),
record(&[(
"children",
encode_rows(&[
record(&[("label", Value::from("b"))]),
record(&[("label", Value::from("c"))]),
]),
)]),
])
.await
.unwrap();
let stored_parents = parents.read(None, None).try_collect::<Vec<_>>().await.unwrap();
assert_eq!(stored_parents.len(), 2);
assert_eq!(children.read(None, None).try_collect::<Vec<_>>().await.unwrap().len(), 3);
let mut created_ids: Vec<_> = ids.iter().map(|row| num(row, "id")).collect();
let mut stored_ids: Vec<_> = stored_parents.iter().map(|row| num(row, "id")).collect();
created_ids.sort();
stored_ids.sort();
assert_eq!(created_ids, stored_ids);
}
fn id_column() -> SchemaColumn {
SchemaColumn {
is_id: true,
..SchemaColumn::new("integer")
}
}
fn named_address_repo(connection: Arc<dyn Connection>) -> CRUDRepository {
let mut options = CRUDRepositoryOptions::default();
options.entity = "Address".into();
options.ids = vec!["Id".into()];
options.property_mappings = Some(Arc::new(columns()));
options.read_properties = vec![
"Id".into(),
"ContactId".into(),
"Line1".into(),
"AddressTypeId".into(),
];
options.create_properties = Some(vec!["ContactId".into(), "Line1".into()]);
options.schema = Schema::new(std::collections::HashMap::from([
("Id".into(), id_column()),
("ContactId".into(), SchemaColumn::new("integer")),
("Line1".into(), SchemaColumn::new("string")),
("AddressTypeName".into(), SchemaColumn::new("string")),
]));
CRUDRepository::<Row>::new(connection, options).unwrap()
}
async fn schema_graph() -> (CRUDRepository, Arc<CRUDRepository>) {
let connection = setup(&mut []).await;
let sources = Arc::new(mapped_crud(connection.clone(), "ContactSource"));
let types = Arc::new(mapped_crud(connection.clone(), "AddressType"));
sources
.create(record(&[("Name", Value::from("import"))]))
.await
.unwrap();
types
.create(record(&[("Name", Value::from("Home"))]))
.await
.unwrap();
let mut addresses = named_address_repo(connection.clone());
addresses.add_enriched(
Enrichment::new(
"AddressTypeName",
"AddressTypeId",
"Id",
Some("Name".into()),
EnrichmentPropertyType::Object,
),
types,
None,
);
let settings = Arc::new(mapped_crud(connection.clone(), "Setting"));
let mut contacts = {
let mut options = CRUDRepositoryOptions::default();
options.entity = "Contact".into();
options.ids = vec!["Id".into()];
options.property_mappings = Some(Arc::new(columns()));
options.read_properties = vec!["Id".into(), "ContactSourceId".into()];
options.create_properties = Some(vec![]);
options.schema = Schema::new(std::collections::HashMap::from([
("Id".into(), id_column()),
("ContactSourceName".into(), SchemaColumn::new("string")),
]));
CRUDRepository::<Row>::new(connection, options).unwrap()
};
let address_repo = Arc::new(addresses);
contacts.add_enriched(
Enrichment::new(
"Addresses",
"Id",
"ContactId",
None,
EnrichmentPropertyType::List,
),
address_repo.clone(),
None,
);
contacts.add_enriched(
Enrichment::new(
"Settings",
"Id",
"ContactId",
Some("Value".into()),
EnrichmentPropertyType::Dictionary,
)
.with_dictionary_key("Name"),
settings,
None,
);
contacts.add_enriched(
Enrichment::new(
"ContactSourceName",
"ContactSourceId",
"Id",
Some("Name".into()),
EnrichmentPropertyType::Object,
),
sources,
None,
);
(contacts, address_repo)
}
#[tokio::test]
async fn create_resolves_address_type_name_without_address_type_id_in_schema() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "AddressType"));
let mut addresses = named_address_repo(connection);
let home = types
.create(record(&[("Name", Value::from("Home"))]))
.await
.unwrap();
addresses.add_enriched(
Enrichment::new(
"AddressTypeName",
"AddressTypeId",
"Id",
Some("Name".into()),
EnrichmentPropertyType::Object,
),
types,
None,
);
let created = addresses
.create(record(&[
("Line1", Value::from("1 Main")),
("AddressTypeName", Value::from("Home")),
]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
let loaded = addresses.read_single(None, None).await.unwrap().unwrap();
assert_eq!(loaded["Line1"], Value::from("1 Main"));
assert_eq!(loaded["AddressTypeName"], Value::from("Home"));
assert_eq!(loaded["AddressTypeId"], home["Id"]);
}
#[tokio::test]
async fn create_keeps_address_type_id_when_name_is_empty() {
let connection = setup(&mut []).await;
let types = Arc::new(mapped_crud(connection.clone(), "AddressType"));
let mut addresses = named_address_repo(connection);
let home = types
.create(record(&[("Name", Value::from("Home"))]))
.await
.unwrap();
addresses.add_enriched(
Enrichment::new(
"AddressTypeName",
"AddressTypeId",
"Id",
Some("Name".into()),
EnrichmentPropertyType::Object,
),
types,
None,
);
let created = addresses
.create(record(&[
("Line1", Value::from("1 Main")),
("AddressTypeId", home["Id"].clone()),
]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
assert_eq!(
addresses.read_single(None, None).await.unwrap().unwrap()["AddressTypeId"],
home["Id"]
);
}
#[tokio::test]
async fn create_contact_stamps_child_contact_id_and_resolves_names() {
let (contacts, addresses) = schema_graph().await;
let created = contacts
.create(record(&[
("ContactSourceName", Value::from("import")),
(
"Addresses",
encode_rows(&[record(&[
("Line1", Value::from("1 Main")),
("AddressTypeName", Value::from("Home")),
])]),
),
(
"Settings",
encode_rows(&[record(&[
("Name", Value::from("theme")),
("Value", Value::from("dark")),
])]),
),
]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
let loaded = contacts.read_single(None, None).await.unwrap().unwrap();
assert_eq!(loaded["ContactSourceName"], Value::from("import"));
let stored = addresses
.read(None, None)
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(stored.len(), 1);
assert_eq!(stored[0]["Line1"], Value::from("1 Main"));
assert_eq!(stored[0]["AddressTypeName"], Value::from("Home"));
assert_eq!(stored[0]["ContactId"], created["Id"]);
}
#[tokio::test]
async fn create_unknown_contact_source_name_throws() {
let (contacts, _) = schema_graph().await;
let error = contacts
.create(record(&[("ContactSourceName", Value::from("Nope"))]))
.await
.unwrap_err();
assert!(error.to_string().contains("ContactSourceName"));
assert!(error.to_string().contains("Nope"));
}
#[tokio::test]
async fn create_missing_contact_source_name_throws() {
let (contacts, _) = schema_graph().await;
let error = contacts.create(Row::new()).await.unwrap_err();
assert!(error.to_string().contains("ContactSourceName"));
}
#[tokio::test]
async fn create_unknown_nested_address_type_name_throws() {
let (contacts, _) = schema_graph().await;
let error = contacts
.create(record(&[
("ContactSourceName", Value::from("import")),
(
"Addresses",
encode_rows(&[record(&[
("Line1", Value::from("1 Main")),
("AddressTypeName", Value::from("Nope")),
])]),
),
]))
.await
.unwrap_err();
assert!(error.to_string().contains("AddressTypeName"));
assert!(error.to_string().contains("Nope"));
}
#[tokio::test]
async fn create_keeps_unknown_contact_source_id_when_name_is_empty() {
let (contacts, _) = schema_graph().await;
let created = contacts
.create(record(&[("ContactSourceId", Value::Int(99))]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
assert_eq!(
contacts.read_single(None, None).await.unwrap().unwrap()["ContactSourceId"],
Value::Int(99)
);
}
#[tokio::test]
async fn create_writes_biginteger_id() {
let connection = setup(&mut []).await;
let mut options = CRUDRepositoryOptions::default();
options.entity = "BigId".into();
options.ids = vec!["Id".into()];
options.schema = Schema::new(std::collections::HashMap::from([
(
"Id".into(),
SchemaColumn {
type_name: "biginteger".into(),
is_id: true,
..SchemaColumn::new("biginteger")
},
),
("Name".into(), SchemaColumn::new("string")),
]));
let repo = CRUDRepository::<Row>::new(connection, options).unwrap();
let created = repo
.create(record(&[("Name", Value::from("Ada"))]))
.await
.unwrap();
assert!(num(&created, "Id") > 0);
assert_eq!(
repo.read_single(None, None).await.unwrap().unwrap()["Name"],
Value::from("Ada")
);
}
fn column_list(sql: &str) -> Vec<&str> {
let open = sql.find('(').unwrap_or(0);
let close = sql[open + 1..]
.find(')')
.map(|index| open + 1 + index)
.unwrap_or(sql.len());
sql[open + 1..close]
.split(", ")
.filter(|name| !name.is_empty())
.collect()
}
struct RecordingConnection {
inner: Arc<dyn Connection>,
texts: Arc<Mutex<Vec<String>>>,
}
impl RecordingConnection {
fn new(inner: Arc<dyn Connection>, texts: Arc<Mutex<Vec<String>>>) -> Self {
Self { inner, texts }
}
}
#[async_trait]
impl Connection for RecordingConnection {
fn as_any(&self) -> &dyn Any {
self
}
fn provider(&self) -> &str {
self.inner.provider()
}
fn is_open(&self) -> bool {
self.inner.is_open()
}
fn create_command(&self) -> Arc<dyn Command> {
Arc::new(RecordingCommand {
inner: self.inner.create_command(),
texts: Arc::clone(&self.texts),
})
}
async fn open(&self) -> Result<()> {
self.inner.open().await
}
async fn close(&self) -> Result<()> {
self.inner.close().await
}
async fn start_transaction(
&self,
isolation: Option<IsolationLevel>,
) -> Result<Arc<dyn Transaction>> {
self.inner.start_transaction(isolation).await
}
}
struct RecordingCommand {
inner: Arc<dyn Command>,
texts: Arc<Mutex<Vec<String>>>,
}
#[async_trait]
impl Command for RecordingCommand {
fn connection(&self) -> Arc<dyn Connection> {
self.inner.connection()
}
fn add_middleware(&self, middleware: Arc<dyn CommandMiddleware>) {
self.inner.add_middleware(middleware);
}
async fn execute_reader_native(
&self,
text: &Native,
parameters: Option<&[Parameter]>,
) -> Result<Box<dyn Reader>> {
self.inner.execute_reader_native(text, parameters).await
}
async fn execute_native(
&self,
text: &Native,
parameters: Option<&[Parameter]>,
) -> Result<Response> {
self.inner.execute_native(text, parameters).await
}
async fn execute_statement(
&self,
statement: &dyn Statement,
parameters: Option<&[Parameter]>,
) -> Result<Response> {
self.texts
.lock()
.expect("texts")
.push(DefaultStatementTextRenderer::instance().render(statement));
self.inner.execute_statement(statement, parameters).await
}
async fn execute_reader_statement(
&self,
statement: &dyn Statement,
parameters: Option<&[Parameter]>,
) -> Result<Box<dyn Reader>> {
self.inner
.execute_reader_statement(statement, parameters)
.await
}
}