use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::fmt::Display;
use std::str::FromStr;
use std::sync::{Arc, Mutex, OnceLock};
use crate::data::query::{
CursorData, DeleteQueryData, NoUser, PageResult, QueryData, RepositoryOptions,
};
use crate::data::types_extra::ChangeResultModel;
use crate::data::{BaseAuditableEntity, BaseEntity};
use chrono::Utc;
use sea_orm::entity::prelude::DateTimeWithTimeZone;
use sea_orm::sea_query::{Alias, Expr, ExprTrait};
use sea_orm::{
ActiveModelTrait, ColumnTrait, DatabaseConnection, DatabaseTransaction, DbErr, EntityTrait,
IntoActiveModel, Iterable, ModelTrait, PaginatorTrait, PrimaryKeyToColumn, PrimaryKeyTrait,
QueryFilter, QuerySelect, TransactionTrait, Value, entity::EntityLoaderTrait,
};
use serde::Serialize;
use uuid::Uuid;
static REGISTRY: OnceLock<Mutex<HashMap<TypeId, Arc<dyn Any + Send + Sync>>>> = OnceLock::new();
fn registry() -> &'static Mutex<HashMap<TypeId, Arc<dyn Any + Send + Sync>>> {
REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}
fn with_cursor<E>(
select: sea_orm::Select<E>,
cursor: &Option<String>,
order_asc: Option<bool>,
) -> sea_orm::Select<E>
where
E: EntityTrait,
{
if let Some(s) = cursor {
if let Some(data) = CursorData::decode(s) {
let is_asc = order_asc.unwrap_or(true);
return if is_asc {
select.filter(Expr::col(Alias::new("id")).gt(data.cursor))
} else {
select.filter(Expr::col(Alias::new("id")).lt(data.cursor))
};
}
}
select
}
fn with_cursor_delete<E>(
delete: sea_orm::DeleteMany<E>,
cursor: &Option<String>,
) -> sea_orm::DeleteMany<E>
where
E: EntityTrait,
{
if let Some(s) = cursor {
if let Some(data) = CursorData::decode(s) {
return delete.filter(Expr::col(Alias::new("id")).gt(data.cursor));
}
}
delete
}
fn with_cursor_loader<E, L>(loader: L, cursor: &Option<String>, order_asc: Option<bool>) -> L
where
E: EntityTrait,
L: EntityLoaderTrait<E>,
{
if let Some(s) = cursor {
if let Some(data) = CursorData::decode(s) {
let is_asc = order_asc.unwrap_or(true);
return if is_asc {
loader.filter(Expr::col(Alias::new("id")).gt(data.cursor))
} else {
loader.filter(Expr::col(Alias::new("id")).lt(data.cursor))
};
}
}
loader
}
fn model_id<E, M>(model: &M) -> i64
where
E: EntityTrait,
M: ModelTrait<Entity = E>,
{
match model.get(E::PrimaryKey::iter().next().unwrap().into_column()) {
Value::BigInt(Some(id)) => id,
Value::Int(Some(id)) => i64::from(id),
value => panic!("repository cursor requires an integer primary key, got {value:?}"),
}
}
fn opts_txn<U>(opts: Option<&RepositoryOptions<U>>) -> Option<Arc<DatabaseTransaction>>
where
U: BaseEntity + Clone + Send + Sync + 'static,
{
opts.and_then(|o| o.txn.clone())
}
fn pick_txn<U>(
opts: Option<&RepositoryOptions<U>>,
qd_opts_txn: Option<Arc<DatabaseTransaction>>,
) -> Option<Arc<DatabaseTransaction>>
where
U: BaseEntity + Clone + Send + Sync + 'static,
{
opts_txn(opts).or(qd_opts_txn)
}
fn try_set<A>(active: &mut A, col_name: &str, value: Value)
where
A: ActiveModelTrait,
A::Entity: EntityTrait,
<A::Entity as EntityTrait>::Column: ColumnTrait + FromStr,
<<A::Entity as EntityTrait>::Column as FromStr>::Err: std::fmt::Debug,
{
if let Ok(col) = <A::Entity as EntityTrait>::Column::from_str(col_name) {
active.set(col, value);
}
}
fn apply_base_create<A, U>(active: &mut A, opts: &RepositoryOptions<U>)
where
A: ActiveModelTrait,
A::Entity: EntityTrait,
<A::Entity as EntityTrait>::Column: ColumnTrait + FromStr,
<<A::Entity as EntityTrait>::Column as FromStr>::Err: std::fmt::Debug,
U: BaseEntity + Clone + Send + Sync + 'static,
{
let now: DateTimeWithTimeZone = Utc::now().into();
let uid = Uuid::new_v4();
try_set(active, "uid", Value::Uuid(Some(uid)));
try_set(active, "createdAt", Value::ChronoDateTimeWithTimeZone(Some(now)));
try_set(active, "updatedAt", Value::ChronoDateTimeWithTimeZone(Some(now)));
let user_id_opt = opts.user_id.or_else(|| opts.user.as_ref().map(|u| u.id()));
if let Some(uid) = user_id_opt {
try_set(active, "createdById", Value::BigInt(Some(uid)));
try_set(active, "updatedById", Value::BigInt(Some(uid)));
}
}
fn apply_base_update<A, U>(active: &mut A, opts: &RepositoryOptions<U>)
where
A: ActiveModelTrait,
A::Entity: EntityTrait,
<A::Entity as EntityTrait>::Column: ColumnTrait + FromStr,
<<A::Entity as EntityTrait>::Column as FromStr>::Err: std::fmt::Debug,
U: BaseEntity + Clone + Send + Sync + 'static,
{
let now: DateTimeWithTimeZone = Utc::now().into();
try_set(active, "updatedAt", Value::ChronoDateTimeWithTimeZone(Some(now)));
let user_id_opt = opts.user_id.or_else(|| opts.user.as_ref().map(|u| u.id()));
if let Some(uid) = user_id_opt {
try_set(active, "updatedById", Value::BigInt(Some(uid)));
}
}
#[derive(Clone)]
pub struct PersistentRepository<E>
where
E: EntityTrait,
E::ModelEx: BaseEntity + Send + Sync,
<E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
{
pub db: DatabaseConnection,
_marker: std::marker::PhantomData<E>,
}
impl<E> PersistentRepository<E>
where
E: EntityTrait,
E::ModelEx: BaseEntity + Send + Sync + Clone + Serialize,
E::Model: Into<E::ModelEx> + Send + Sync,
<E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
pub fn new(db: DatabaseConnection) -> Self {
Self {
db,
_marker: std::marker::PhantomData,
}
}
pub fn db(&self) -> &DatabaseConnection {
&self.db
}
pub fn initialize(db: &DatabaseConnection) -> Arc<Self> {
let type_id = TypeId::of::<E>();
let mut reg = registry().lock().unwrap();
if reg.contains_key(&type_id) {
panic!(
"Repository for {} has already been initialized",
std::any::type_name::<E>()
);
}
let repo = Arc::new(Self::new(db.clone()));
reg.insert(type_id, repo.clone() as Arc<dyn Any + Send + Sync>);
repo
}
pub fn find() -> Arc<Self> {
Self::try_find().unwrap_or_else(|| panic!("Repository for {} has not been initialized — call PersistentRepository::<{}>::initialize(db) first", std::any::type_name::<E>(), std::any::type_name::<E>()))
}
pub fn try_find() -> Option<Arc<Self>> {
let reg = registry().lock().unwrap();
reg.get(&TypeId::of::<E>())
.and_then(|arc| Arc::downcast::<Self>(arc.clone()).ok())
}
pub fn new_ephemeral(db: DatabaseConnection) -> Self {
Self::new(db)
}
pub fn start(&self) -> QueryData<E>
where
E::ModelEx: BaseEntity,
{
QueryData::new(<E::ModelEx as BaseEntity>::load())
}
pub fn start_delete(&self) -> DeleteQueryData<E> {
DeleteQueryData::new(E::delete_many())
}
pub fn is_owner<U>(&self, model: &E::ModelEx, user: &U) -> bool
where
U: BaseEntity,
{
model.id() == user.id()
}
pub fn is_owner_auditable<U>(&self, model: &E::ModelEx, user: &U) -> bool
where
U: BaseEntity + Clone + Send + Sync,
E::ModelEx: BaseAuditableEntity<User = U>,
{
model.created_by().id() == user.id()
}
pub fn is_accessible<U>(&self, model: &E::ModelEx, user: &U) -> bool
where
U: BaseEntity,
{
self.is_owner(model, user)
}
pub fn is_accessible_auditable<U>(&self, model: &E::ModelEx, user: &U) -> bool
where
U: BaseEntity + Clone + Send + Sync,
E::ModelEx: BaseAuditableEntity<User = U>,
{
model.created_by().id() == user.id() || model.updated_by().id() == user.id()
}
pub async fn transaction<F, T, EType>(&self, f: F) -> Result<T, EType>
where
EType: From<DbErr> + Display + Send,
F: FnOnce(
Arc<DatabaseTransaction>,
) -> futures::future::BoxFuture<'static, Result<T, EType>>
+ Send,
T: Send,
{
let txn = Arc::new(self.db.begin().await?);
let res = f(txn.clone()).await?;
let owned = Arc::try_unwrap(txn).map_err(|_| {
DbErr::Custom(
"transaction handle still shared after closure; refusing to commit".to_string(),
)
})?;
owned.commit().await?;
Ok(res)
}
pub async fn transaction_with_opts<U, F, T, EType>(
&self,
opts: RepositoryOptions<U>,
f: F,
) -> Result<T, EType>
where
EType: From<DbErr> + Display + Send,
U: BaseEntity + Clone + Send + Sync + 'static,
F: FnOnce(
Arc<DatabaseTransaction>,
) -> futures::future::BoxFuture<'static, Result<T, EType>>
+ Send,
T: Send,
{
if let Some(txn) = opts.txn {
return f(txn).await;
}
self.transaction(f).await
}
pub async fn get_count<U>(
&self,
filter: Option<QueryData<E>>,
opts: Option<RepositoryOptions<U>>,
) -> Result<u64, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: BaseEntity,
{
let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
let txn = pick_txn(
opts.as_ref(),
filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
);
let (loader, query_cursor, order) = match filter {
Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
None => (<E::ModelEx as BaseEntity>::load(), None, None),
};
let loader = with_cursor_loader(loader, &cursor.or(query_cursor), order);
if let Some(txn) = txn.as_ref() {
loader.num_items(txn.as_ref(), 0).await
} else {
loader.num_items(&self.db, 0).await
}
}
pub async fn get_count_txn(
&self,
filter: Option<QueryData<E>>,
txn: &DatabaseTransaction,
) -> Result<u64, DbErr>
where
E::ModelEx: BaseEntity,
{
let (loader, cursor, order) = match filter {
Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
None => (<E::ModelEx as BaseEntity>::load(), None, None),
};
with_cursor_loader(loader, &cursor, order)
.num_items(txn, 0)
.await
}
pub async fn count(&self, filter: Option<QueryData<E>>) -> Result<u64, DbErr>
where
E::ModelEx: BaseEntity,
{
self.get_count::<NoUser>(filter, None).await
}
pub async fn get_paginated_view<U>(
&self,
filter: Option<QueryData<E>>,
opts: RepositoryOptions<U>,
) -> Result<PageResult<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: BaseEntity,
{
let order_asc = filter.as_ref().and_then(|qd| qd.order_asc);
let txn = opts
.txn
.clone()
.or_else(|| filter.as_ref().and_then(|qd| qd.opts.txn.clone()));
let loader = filter
.map(|qd| qd.loader)
.unwrap_or_else(<E::ModelEx as BaseEntity>::load);
let total_loader = loader.clone();
let total = if let Some(txn) = txn.as_ref() {
total_loader.num_items(txn.as_ref(), 0).await? as i64
} else {
total_loader.num_items(&self.db, 0).await? as i64
};
let sel = with_cursor_loader(loader, &opts.cursor, order_asc);
let limit = opts.limit;
let rows: Vec<E::ModelEx> = if let Some(txn) = txn.as_ref() {
sel.fetch(txn.as_ref(), 0, (limit + 1) as u64).await?
} else {
sel.fetch(&self.db, 0, (limit + 1) as u64).await?
};
let has_next = rows.len() > limit;
let mut data = rows;
if has_next {
data.truncate(limit);
}
let next_cursor = if has_next {
Some(
CursorData {
limit,
cursor: model_id::<E, _>(data.last().unwrap()),
}
.encode(),
)
} else {
None
};
Ok(PageResult {
data,
has_next,
cursor: opts.cursor.clone(),
next_cursor,
total,
limit,
})
}
pub async fn paginated(
&self,
select: sea_orm::Select<E>,
opts: RepositoryOptions<NoUser>,
) -> Result<PageResult<E::ModelEx>, DbErr>
where
E::ModelEx: BaseEntity,
E::Model: Into<E::ModelEx> + Send + Sync,
{
let limit = opts.limit;
let cursor = opts.cursor.clone();
let txn = opts.txn.clone();
let mut count_sel = select.clone();
if opts.distinct {
count_sel = count_sel.distinct();
}
let total = if let Some(txn) = txn.as_ref() {
count_sel.paginate(txn.as_ref(), 1).num_items().await? as i64
} else {
count_sel.paginate(&self.db, 1).num_items().await? as i64
};
let mut sel = with_cursor(select, &cursor, None);
if opts.distinct {
sel = sel.distinct();
}
sel = sel.limit((limit + 1) as u64);
let rows: Vec<E::ModelEx> = if let Some(txn) = txn.as_ref() {
sel.all(txn.as_ref())
.await?
.into_iter()
.map(Into::into)
.collect()
} else {
sel.all(&self.db)
.await?
.into_iter()
.map(Into::into)
.collect()
};
let has_next = rows.len() > limit;
let mut data = rows;
if has_next {
data.truncate(limit);
}
let next_cursor = if has_next {
Some(
CursorData {
limit,
cursor: model_id::<E, _>(data.last().unwrap()),
}
.encode(),
)
} else {
None
};
Ok(PageResult {
data,
has_next,
cursor: opts.cursor,
next_cursor,
total,
limit: opts.limit,
})
}
pub async fn get_all<U>(
&self,
filter: Option<QueryData<E>>,
opts: Option<RepositoryOptions<U>>,
) -> Result<Vec<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: BaseEntity,
{
let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
let txn = pick_txn(
opts.as_ref(),
filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
);
let (loader, qd_cursor, order) = match filter {
Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
None => (<E::ModelEx as BaseEntity>::load(), None, None),
};
let cursor = cursor.or(qd_cursor);
let loader = with_cursor_loader(loader, &cursor, order);
if let Some(txn) = txn.as_ref() {
loader.fetch(txn.as_ref(), 0, 0).await
} else {
loader.fetch(&self.db, 0, 0).await
}
}
pub async fn get_many<U>(
&self,
filter: Option<QueryData<E>>,
opts: RepositoryOptions<U>,
) -> Result<Vec<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: BaseEntity,
{
let cursor = opts.cursor.clone();
let txn = opts
.txn
.clone()
.or_else(|| filter.as_ref().and_then(|qd| qd.opts.txn.clone()));
let (loader, qd_cursor, order) = match filter {
Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
None => (<E::ModelEx as BaseEntity>::load(), None, None),
};
let cursor = if cursor.is_some() { cursor } else { qd_cursor };
let loader = with_cursor_loader(loader, &cursor, order);
if let Some(txn) = txn.as_ref() {
loader.fetch(txn.as_ref(), 0, opts.limit as u64).await
} else {
loader.fetch(&self.db, 0, opts.limit as u64).await
}
}
pub async fn find_many<U>(
&self,
filter: Option<QueryData<E>>,
opts: RepositoryOptions<U>,
) -> Result<Vec<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: BaseEntity,
{
self.get_many(filter, opts).await
}
pub async fn find_all(&self, filter: Option<QueryData<E>>) -> Result<Vec<E::ModelEx>, DbErr>
where
E::ModelEx: BaseEntity,
{
self.get_all::<NoUser>(filter, None).await
}
pub async fn get_distinct_rows<U>(
&self,
filter: Option<QueryData<E>>,
opts: RepositoryOptions<U>,
) -> Result<Vec<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: BaseEntity,
{
let cursor = opts.cursor.clone();
let txn = opts
.txn
.clone()
.or_else(|| filter.as_ref().and_then(|qd| qd.opts.txn.clone()));
let (loader, qd_cursor, order) = match filter {
Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
None => (<E::ModelEx as BaseEntity>::load(), None, None),
};
let loader = with_cursor_loader(loader, &cursor.or(qd_cursor), order);
if let Some(txn) = txn.as_ref() {
loader.fetch(txn.as_ref(), 0, opts.limit as u64).await
} else {
loader.fetch(&self.db, 0, opts.limit as u64).await
}
}
pub async fn get_one<U>(
&self,
filter: Option<QueryData<E>>,
opts: Option<RepositoryOptions<U>>,
) -> Result<Option<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: BaseEntity,
{
let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
let txn = pick_txn(
opts.as_ref(),
filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
);
let (loader, qd_cursor, order) = match filter {
Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
None => (<E::ModelEx as BaseEntity>::load(), None, None),
};
let loader = with_cursor_loader(loader, &cursor.or(qd_cursor), order);
if let Some(txn) = txn.as_ref() {
Ok(loader.fetch(txn.as_ref(), 0, 1).await?.into_iter().next())
} else {
Ok(loader.fetch(&self.db, 0, 1).await?.into_iter().next())
}
}
pub async fn find_one(&self, filter: Option<QueryData<E>>) -> Result<Option<E::ModelEx>, DbErr>
where
E::ModelEx: BaseEntity,
{
self.get_one::<NoUser>(filter, None).await
}
pub async fn get_by_id(&self, id: i64) -> Result<Option<E::ModelEx>, DbErr>
where
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::ModelEx: BaseEntity,
<E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
{
<E::ModelEx as BaseEntity>::load()
.filter_by_id(id)
.fetch(&self.db, 0, 1)
.await
.map(|mut rows| rows.pop())
}
pub async fn find_by_id(&self, id: i64) -> Result<Option<E::ModelEx>, DbErr>
where
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::ModelEx: BaseEntity,
{
self.get_by_id(id).await
}
pub async fn get_by_id_with_opts<U>(
&self,
id: i64,
opts: RepositoryOptions<U>,
) -> Result<Option<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::ModelEx: BaseEntity,
{
let loader = <E::ModelEx as BaseEntity>::load().filter_by_id(id);
if let Some(txn) = opts.txn.as_ref() {
Ok(loader.fetch(txn.as_ref(), 0, 1).await?.into_iter().next())
} else {
Ok(loader.fetch(&self.db, 0, 1).await?.into_iter().next())
}
}
pub async fn create_one<A, U>(
&self,
mut active: A,
opts: Option<RepositoryOptions<U>>,
) -> Result<E::ModelEx, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::Model: IntoActiveModel<A>,
E::ModelEx: BaseEntity,
U: BaseEntity + Clone + Send + Sync + 'static,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
if let Some(o) = opts.as_ref() {
apply_base_create(&mut active, o);
} else {
let empty: RepositoryOptions<U> = RepositoryOptions::default();
apply_base_create(&mut active, &empty);
}
if let Some(txn) = opts_txn(opts.as_ref()).as_ref() {
active.insert(txn.as_ref()).await.map(Into::into)
} else {
active.insert(&self.db).await.map(Into::into)
}
}
pub async fn create_one_simple<A>(&self, active: A) -> Result<E::ModelEx, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::Model: IntoActiveModel<A>,
E::ModelEx: BaseEntity,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
self.create_one::<A, NoUser>(active, None).await
}
pub async fn create_one_with_opts<A, U>(
&self,
active: A,
opts: RepositoryOptions<U>,
) -> Result<E::ModelEx, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::Model: IntoActiveModel<A>,
E::ModelEx: BaseEntity,
U: BaseEntity + Clone + Send + Sync + 'static,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
self.create_one(active, Some(opts)).await
}
pub async fn create_many<A, U>(
&self,
mut actives: Vec<A>,
opts: Option<RepositoryOptions<U>>,
) -> Result<Vec<E::ModelEx>, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::ModelEx: BaseEntity,
E::Model: IntoActiveModel<A>,
U: BaseEntity + Clone + Send + Sync + 'static,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let mut out = Vec::with_capacity(actives.len());
for active in &mut actives {
if let Some(o) = opts.as_ref() {
apply_base_create(active, o);
} else {
let empty: RepositoryOptions<U> = RepositoryOptions::default();
apply_base_create(active, &empty);
}
}
let txn = opts_txn(opts.as_ref());
for a in actives {
if let Some(txn) = txn.as_ref() {
out.push(a.insert(txn.as_ref()).await?.into());
} else {
out.push(a.insert(&self.db).await?.into());
}
}
Ok(out)
}
pub async fn create_many_simple<A>(&self, actives: Vec<A>) -> Result<Vec<E::ModelEx>, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::ModelEx: BaseEntity,
E::Model: IntoActiveModel<A>,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
self.create_many::<A, NoUser>(actives, None).await
}
pub async fn create_many_with_opts<A, U>(
&self,
actives: Vec<A>,
opts: RepositoryOptions<U>,
) -> Result<Vec<E::ModelEx>, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::ModelEx: BaseEntity,
E::Model: IntoActiveModel<A>,
U: BaseEntity + Clone + Send + Sync + 'static,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
self.create_many(actives, Some(opts)).await
}
pub async fn upsert_one<A>(&self, mut active: A) -> Result<E::ModelEx, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::Model: IntoActiveModel<A> + Clone,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_create(&mut active, &empty);
active.insert(&self.db).await.map(Into::into)
}
pub async fn upsert_many<A>(&self, actives: Vec<A>) -> Result<Vec<E::ModelEx>, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::ModelEx: BaseEntity,
E::Model: IntoActiveModel<A> + Clone,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let mut out = Vec::with_capacity(actives.len());
for a in actives {
out.push(self.upsert_one(a).await?);
}
Ok(out)
}
pub async fn upsert_one_with_id<A, F>(
&self,
mut active: A,
id_fn: F,
) -> Result<E::ModelEx, DbErr>
where
A: ActiveModelTrait<Entity = E> + Send,
A: sea_orm::ActiveModelBehavior + Send,
E::Model: IntoActiveModel<A> + Clone + BaseEntity,
F: Fn(&A) -> Option<i64> + Send,
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
if let Some(id) = id_fn(&active) {
if self.get_by_id(id).await?.is_some() {
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_update(&mut active, &empty);
return active.update(&self.db).await.map(Into::into);
}
}
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_create(&mut active, &empty);
active.insert(&self.db).await.map(Into::into)
}
pub async fn update_many<F, U>(
&self,
filter: Option<QueryData<E>>,
mut effector: F,
opts: RepositoryOptions<U>,
) -> Result<ChangeResultModel<E::ModelEx>, DbErr>
where
F: FnMut(&mut E::Model) -> bool + Send,
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: Into<E::Model> + BaseEntity,
E::Model: IntoActiveModel<E::ActiveModel> + Clone,
E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let txn = opts.txn.clone();
let items = self.get_many(filter, opts.clone()).await?;
let mut affected = Vec::new();
for model in items {
let mut model: E::Model = model.into();
if effector(&mut model) {
let mut am: E::ActiveModel = model.clone().into_active_model();
apply_base_update(&mut am, &opts);
if let Some(txn) = txn.as_ref() {
am.update(txn.as_ref()).await?;
} else {
am.update(&self.db).await?;
}
affected.push(model.into());
}
}
Ok(ChangeResultModel::new(affected))
}
pub async fn find_update_many<F, U>(
&self,
filter: Option<QueryData<E>>,
effector: F,
opts: RepositoryOptions<U>,
) -> Result<ChangeResultModel<E::ModelEx>, DbErr>
where
F: FnMut(&mut E::Model) -> bool + Send,
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: Into<E::Model> + BaseEntity,
E::Model: IntoActiveModel<E::ActiveModel> + Clone,
E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
self.update_many(filter, effector, opts).await
}
pub async fn update_one<F>(
&self,
filter: Option<QueryData<E>>,
mut effector: F,
) -> Result<Option<E::ModelEx>, DbErr>
where
F: FnMut(&mut E::Model) -> bool + Send,
E::ModelEx: Into<E::Model> + BaseEntity,
E::Model: IntoActiveModel<E::ActiveModel> + Clone ,
E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let Some(model) = self.get_one::<NoUser>(filter, None).await? else {
return Ok(None);
};
let mut model: E::Model = model.into();
if !effector(&mut model) {
return Ok(Some(model.into()));
}
let mut am: E::ActiveModel = model.clone().into_active_model();
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_update(&mut am, &empty);
let updated = am.update(&self.db).await?;
Ok(Some(updated.into()))
}
pub async fn update_by_id<F>(
&self,
id: i64,
mut effector: F,
) -> Result<Option<E::ModelEx>, DbErr>
where
F: FnMut(&mut E::Model) -> bool + Send,
E::ModelEx: Into<E::Model> + BaseEntity,
E::Model: IntoActiveModel<E::ActiveModel> + Clone,
E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let Some(model) = self.get_by_id(id).await? else {
return Ok(None);
};
let mut model: E::Model = model.into();
if !effector(&mut model) {
return Ok(Some(model.into()));
}
let mut am: E::ActiveModel = model.clone().into_active_model();
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_update(&mut am, &empty);
let updated = am.update(&self.db).await?;
Ok(Some(updated.into()))
}
pub async fn update_by_id_with_active<F>(
&self,
id: i64,
mut effector: F,
) -> Result<Option<E::ModelEx>, DbErr>
where
F: FnMut(&mut E::ActiveModel) -> bool + Send,
E::ModelEx: Into<E::Model> + BaseEntity,
E::Model: IntoActiveModel<E::ActiveModel> + Clone,
E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let Some(model) = self.get_by_id(id).await? else {
return Ok(None);
};
let model: E::Model = model.into();
let mut am: E::ActiveModel = model.into_active_model();
if !effector(&mut am) {
return self.get_by_id(id).await;
}
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_update(&mut am, &empty);
let updated = am.update(&self.db).await?;
Ok(Some(updated.into()))
}
pub async fn update_one_with_opts<F, U>(
&self,
filter: Option<QueryData<E>>,
mut effector: F,
opts: Option<RepositoryOptions<U>>,
) -> Result<Option<E::ModelEx>, DbErr>
where
F: FnMut(&mut E::Model) -> bool + Send,
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: Into<E::Model>,
E::Model: IntoActiveModel<E::ActiveModel> + Clone,
E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let txn = pick_txn(
opts.as_ref(),
filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
);
let Some(model) = self.get_one(filter, opts.clone()).await? else {
return Ok(None);
};
let mut model: E::Model = model.into();
if !effector(&mut model) {
return Ok(Some(model.into()));
}
let mut am: E::ActiveModel = model.clone().into_active_model();
if let Some(o) = opts.as_ref() {
apply_base_update(&mut am, o);
} else {
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_update(&mut am, &empty);
}
let updated = if let Some(txn) = txn.as_ref() {
am.update(txn.as_ref()).await?
} else {
am.update(&self.db).await?
};
Ok(Some(updated.into()))
}
pub async fn update_by_id_with_opts<F, U>(
&self,
id: i64,
mut effector: F,
opts: Option<RepositoryOptions<U>>,
) -> Result<Option<E::ModelEx>, DbErr>
where
F: FnMut(&mut E::Model) -> bool + Send,
U: BaseEntity + Clone + Send + Sync + 'static,
E::ModelEx: Into<E::Model> + BaseEntity,
E::Model: IntoActiveModel<E::ActiveModel> + Clone,
E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::Column: ColumnTrait + FromStr,
<E::Column as FromStr>::Err: std::fmt::Debug,
{
let txn = opts_txn(opts.as_ref());
let Some(model) = (if let Some(o) = opts.clone() {
self.get_by_id_with_opts(id, o).await?
} else {
self.get_by_id(id).await?
}) else {
return Ok(None);
};
let mut model: E::Model = model.into();
if !effector(&mut model) {
return Ok(Some(model.into()));
}
let mut am: E::ActiveModel = model.clone().into_active_model();
if let Some(o) = opts.as_ref() {
apply_base_update(&mut am, o);
} else {
let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
apply_base_update(&mut am, &empty);
}
let updated = if let Some(txn) = txn.as_ref() {
am.update(txn.as_ref()).await?
} else {
am.update(&self.db).await?
};
Ok(Some(updated.into()))
}
pub async fn delete_many(&self, filter: Option<DeleteQueryData<E>>) -> Result<u64, DbErr> {
self.delete_many_with_opts::<NoUser>(filter, None).await
}
pub async fn delete_many_with_opts<U>(
&self,
filter: Option<DeleteQueryData<E>>,
opts: Option<RepositoryOptions<U>>,
) -> Result<u64, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
{
let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
let txn = opts_txn(opts.as_ref());
let del = filter
.map(|qd| qd.delete)
.unwrap_or_else(|| E::delete_many());
let del = with_cursor_delete(del, &cursor);
let res = if let Some(txn) = txn.as_ref() {
del.exec(txn.as_ref()).await?
} else {
del.exec(&self.db).await?
};
Ok(res.rows_affected)
}
pub async fn remove_many(&self, filter: Option<DeleteQueryData<E>>) -> Result<u64, DbErr> {
self.delete_many(filter).await
}
pub async fn remove_many_with_opts<U>(
&self,
filter: Option<DeleteQueryData<E>>,
opts: Option<RepositoryOptions<U>>,
) -> Result<u64, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
{
self.delete_many_with_opts(filter, opts).await
}
pub async fn delete_one(&self, filter: Option<DeleteQueryData<E>>) -> Result<bool, DbErr> {
self.delete_one_with_opts::<NoUser>(filter, None).await
}
pub async fn delete_one_with_opts<U>(
&self,
filter: Option<DeleteQueryData<E>>,
opts: Option<RepositoryOptions<U>>,
) -> Result<bool, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
{
Ok(self.delete_many_with_opts(filter, opts).await? > 0)
}
pub async fn delete_by_id(&self, id: i64) -> Result<Option<E::ModelEx>, DbErr>
where
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::ModelEx: BaseEntity,
{
let Some(model) = self.get_by_id(id).await? else {
return Ok(None);
};
let pk = <<E::PrimaryKey as PrimaryKeyTrait>::ValueType as From<i64>>::from(id);
E::delete_by_id(pk).exec(&self.db).await?;
Ok(Some(model))
}
pub async fn delete_by_id_with_opts<U>(
&self,
id: i64,
opts: Option<RepositoryOptions<U>>,
) -> Result<Option<E::ModelEx>, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::ModelEx: BaseEntity,
{
let txn = opts_txn(opts.as_ref());
let Some(model) = (if let Some(txn) = txn.as_ref() {
<E::ModelEx as BaseEntity>::load()
.filter_by_id(id)
.fetch(txn.as_ref(), 0, 1)
.await?
.into_iter()
.next()
} else {
self.get_by_id(id).await?
}) else {
return Ok(None);
};
let pk = <<E::PrimaryKey as PrimaryKeyTrait>::ValueType as From<i64>>::from(id);
if let Some(txn) = txn.as_ref() {
E::delete_by_id(pk).exec(txn.as_ref()).await?;
} else {
E::delete_by_id(pk).exec(&self.db).await?;
}
Ok(Some(model))
}
pub async fn delete_by_id_txn(
&self,
id: i64,
txn: &DatabaseTransaction,
) -> Result<Option<E::ModelEx>, DbErr>
where
E::PrimaryKey: PrimaryKeyTrait,
<E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
E::ModelEx: BaseEntity,
{
let pk = <<E::PrimaryKey as PrimaryKeyTrait>::ValueType as From<i64>>::from(id);
let Some(model) = <E::ModelEx as BaseEntity>::load()
.filter_by_id(id)
.fetch(txn, 0, 1)
.await?
.into_iter()
.next()
else {
return Ok(None);
};
E::delete_by_id(pk).exec(txn).await?;
Ok(Some(model))
}
pub async fn delete_many_txn(
&self,
filter: Option<DeleteQueryData<E>>,
txn: &DatabaseTransaction,
) -> Result<u64, DbErr> {
self.delete_many_txn_with_opts::<NoUser>(filter, None, txn)
.await
}
pub async fn delete_many_txn_with_opts<U>(
&self,
filter: Option<DeleteQueryData<E>>,
opts: Option<RepositoryOptions<U>>,
txn: &DatabaseTransaction,
) -> Result<u64, DbErr>
where
U: BaseEntity + Clone + Send + Sync + 'static,
{
let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
let del = filter
.map(|qd| qd.delete)
.unwrap_or_else(|| E::delete_many());
let del = with_cursor_delete(del, &cursor);
if let Some(opts_txn) = opts_txn(opts.as_ref()) {
let res = del.exec(opts_txn.as_ref()).await?;
return Ok(res.rows_affected);
}
let res = del.exec(txn).await?;
Ok(res.rows_affected)
}
}
pub fn new_uid() -> Uuid {
Uuid::new_v4()
}