基本操作

对于每个操作，您需要将相应的特质引入作用域。它们在 charybdis::operations 模块中定义。

插入

• use charybdis::{CachingSession, Insert};
  
  #[tokio::main]
  async fn main() {
    let session: &CachingSession; // init sylla session
    
    // init user
    let user: User = User {
      id,
      email: "[email protected]".to_string(),
      username: "charybdis".to_string(),
      created_at: Utc::now(),
      updated_at: Utc::now(),
      address: Some(
          Address {
              street: "street".to_string(),
              state: "state".to_string(),
              zip: "zip".to_string(),
              country: "country".to_string(),
              city: "city".to_string(),
          }
      ),
    };
  
    // create
    user.insert().execute(&session).await;
  }
  

查找

• 按主键查找

    let user = User {id, ..Default::default()};
    let user = user.find_by_primary_key().execute(&session).await?;
  

• 按分区键查找

    let users =  User {id, ..Default::default()}.find_by_partition_key().execute(&session).await;
  

• 按关联主键查找

  let users = User::find_by_primary_key_value(val: User::PrimaryKey).execute(&session).await;
  

• 可用的查找函数

  use scylla::CachingSession;
  use charybdis::errors::CharybdisError;
  use charybdis::macros::charybdis_model;
  use charybdis::stream::CharybdisModelStream;
  use charybdis::types::{Date, Text, Uuid};
  
  #[charybdis_model(
      table_name = posts,
      partition_keys = [date],
      clustering_keys = [category_id, title],
      global_secondary_indexes = [category_id],
      local_secondary_indexes = [title]
  )]
  pub struct Post {
      pub date: Date,
      pub category_id: Uuid,
      pub title: Text,
  }
  
  impl Post {
      async fn find_various(db_session: &CachingSession) -> Result<(), CharybdisError> {
         let date = Date::default();
         let category_id = Uuid::new_v4();
         let title = Text::default();
      
         let posts: CharybdisModelStream<Post> = Post::find_by_date(date).execute(db_session).await?;
         let posts: CharybdisModelStream<Post> = Post::find_by_date_and_category_id(date, category_id).execute(db_session).await?;
         let posts: Post = Post::find_by_date_and_category_id_and_title(date, category_id, title.clone()).execute(db_session).await?;
      
         let post: Post = Post::find_first_by_date(date).execute(db_session).await?;
         let post: Post = Post::find_first_by_date_and_category_id(date, category_id).execute(db_session).await?;
      
         let post: Option<Post> = Post::maybe_find_first_by_date(date).execute(db_session).await?;
         let post: Option<Post> = Post::maybe_find_first_by_date_and_category_id(date, category_id).execute(db_session).await?;
         let post: Option<Post> = Post::maybe_find_first_by_date_and_category_id_and_title(date, category_id, title.clone()).execute(db_session).await?;
      
         // find by local secondary index
         let posts: CharybdisModelStream<Post> = Post::find_by_date_and_title(date, title.clone()).execute(db_session).await?;
         let post: Post = Post::find_first_by_date_and_title(date, title.clone()).execute(db_session).await?;
         let post: Option<Post> = Post::maybe_find_first_by_date_and_title(date, title.clone()).execute(db_session).await?;
  
        // find by global secondary index
        let posts: CharybdisModelStream<Post> = Post::find_by_category_id(category_id).execute(db_session).await?;
        let post: Post = Post::find_first_by_category_id(category_id).execute(db_session).await?;
        let post: Option<Post> = Post::maybe_find_first_by_category_id(category_id).execute(db_session).await?;
      
        Ok(())
      }
  }
  

• 自定义过滤

  让我们以我们的 Post 模型为例

  #[charybdis_model(
      table_name = posts, 
      partition_keys = [category_id], 
      clustering_keys = [date, title],
      global_secondary_indexes = []
  )]
  pub struct Post {...}
  

  我们自动生成了遵循以下约定的宏 find_<struct_name>! 的 find_post! 宏。它可以用来创建自定义查询。

  以下代码将返回 Post 模型的流，并且查询将在编译时构建为 &'static str。

  // automatically generated macro rule
  let posts = find_post!("category_id in ? AND date > ?", (categor_vec, date))
      .execute(session)
      .await?;
  

  我们还可以使用 find_first_post! 宏来获取单个结果

  let post = find_first_post!("category_id in ? AND date > ? LIMIT 1", (date, categor_vec))
      .execute(session)
      .await?;
  

  如果我们只需要 Query 而不是结果，我们可以使用 find_post_query! 宏

  let query = find_post_query!("date = ? AND category_id in ?", (date, categor_vec));
  

更新

• let user = User::from_json(json);
  
  user.username = "scylla".to_string();
  user.email = "[email protected]";
  
  user.update().execute(&session).await;
  

• 收集

  • 让我们以我们的 User 模型为例

    #[charybdis_model(
        table_name = users,
        partition_keys = [id],
        clustering_keys = [],
    )]
    pub struct User {
        id: Uuid,
        tags: Set<Text>,
        post_ids: List<Uuid>,
    }
    

  • 为每个集合字段生成了 push_to_<field_name> 和 pull_from_<field_name> 方法。

    let user: User;
    
    user.push_tags(vec![tag]).execute(&session).await;
    user.pull_tags(vec![tag]).execute(&session).await;
    
    user.push_post_ids(vec![tag]).execute(&session).await;
    user.pull_post_ids(vec![tag]).execute(&session).await;
    

• 计数器

  • 让我们定义 post_counter 模型

    #[charybdis_model(
        table_name = post_counters,
        partition_keys = [id],
        clustering_keys = [],
    )]
    pub struct PostCounter {
        id: Uuid,
        likes: Counter,
        comments: Counter,
    }
    

  • 我们可以使用 increment_<field_name> 和 decrement_<field_name> 方法来更新计数器字段。

    let post_counter: PostCounter;
    post_counter.increment_likes(1).execute(&session).await;
    post_counter.decrement_likes(1).execute(&session).await;
    
    post_counter.increment_comments(1).execute(&session).await;
    post_counter.decrement_comments(1).execute(&session).await;
    

删除

• let user = User::from_json(json);
  
  user.delete().execute(&session).await;
  

• 宏生成的删除助手

  让我们以我们的 Post 模型为例

  #[charybdis_model(
      table_name = posts,
      partition_keys = [date],
      clustering_keys = [categogry_id, title],
      global_secondary_indexes = [])
  ]
  pub struct Post {
      date: Date,
      category_id: Uuid,
      title: Text,
      id: Uuid,
      ...
  }
  

  我们为从主键生成的最多 3 个字段有宏生成的函数。

  Post::delete_by_date(date: Date).execute(&session).await?;
  Post::delete_by_date_and_category_id(date: Date, category_id: Uuid).execute(&session).await?;
  Post::delete_by_date_and_category_id_and_title(date: Date, category_id: Uuid, title: Text).execute(&session).await?;
  

• 自定义删除查询

  我们可以使用 delete_post! 宏来创建自定义删除查询。

  delete_post!("date = ? AND category_id in ?", (date, category_vec)).execute(&session).await?
  

配置

每个操作都返回 CharybdisQuery，在执行前可以通过方法链进行配置。

let user: User = User::find_by_id(id)
    .consistency(Consistency::One)
    .timeout(Some(Duration::from_secs(5)))
    .execute(&app.session)
    .await?;
    
let result: QueryResult = user.update().consistency(Consistency::One).execute(&session).await?;

支持的配置选项

• 一致性
• 串行一致性
• 时间戳
• 超时
• 页面大小
• 时间戳

批处理

使用 CharybdisModelBatch 操作来执行单个批处理中的多个操作。

• 批量操作

  let users: Vec<User>;
  let batch = User::batch();
  
  // inserts
  batch.append_inserts(users);
  
  // or updates
  batch.append_updates(users);
  
  // or deletes
  batch.append_deletes(users);
  
  batch.execute(&session).await?;
  

• 分块批量操作

  分块批处理操作用于分块处理大量数据。

    let users: Vec<User>;
    let chunk_size = 100;
  
    User::batch().chunked_inserts(&session, users, chunk_size).await?;
    User::batch().chunked_updates(&session, users, chunk_size).await?;
    User::batch().chunked_deletes(&session, users, chunk_size).await?;
  

• 批量配置

  批处理操作可以在执行前通过方法链进行配置。

  let batch = User::batch()
      .consistency(Consistency::One)
      .retry_policy(Some(Arc::new(DefaultRetryPolicy::new())))
      .chunked_inserts(&session, users, 100)
      .await?;
  

  我们也可以使用方法链将操作附加到批处理

  let batch = User::batch()
      .consistency(Consistency::One)
      .retry_policy(Some(Arc::new(DefaultRetryPolicy::new())))
      .append_update(&user_1)
      .append_update(&user_2)
      .execute(data.db_session())
      .await?;
  

• 语句批处理

  我们可以使用批处理语句来批量执行集合操作。

  let batch = User::batch();
  let users: Vec<User>;
  
  for user in users {
      batch.append_statement(User::PUSH_TAGS_QUERY, (vec![tag], user.id));
  }
  
  batch.execute(&session).await;
  

部分模型

• 使用自动生成的 partial_<model>! 宏来在模型字段的子集上运行操作。此宏生成一个结构与原始模型相同的新结构，但只包含提供的字段。宏由 #[charybdis_model] 自动生成。它遵循以下约定 partial_<struct_name>!。

  // auto-generated macro - available in crate::models::user
  partial_user!(UpdateUsernameUser, id, username);
  

  现在我们有了新的结构体 UpdateUsernameUser，它与 User 模型等效，但只包含 id 和 username 字段。

  let mut update_user_username = UpdateUsernameUser {
      id,
      username: "updated_username".to_string(),
  };
  
  update_user_username.update().execute(&session).await?;
  

• 部分模型注意事项

  • partial_<model> 要求在本地模型上使用 #[derive(Default)]
  • partial_<model> 需要在定义中包含完整的键
  • 在 #charybdis_model 宏下方定义的所有派生项都将自动添加到部分模型中。
  • partial_<model> 结构体实现了与原生模型相同的字段属性，因此如果我们原生模型字段上存在 #[serde(rename = "rootId")] 注解，那么它也会出现在部分模型字段上。
  • partial_<model> 应该定义在原生模型相同的文件中，以便它可以重用原生模型所需的导入。

• 作为本地

  如果我们需要在原生模型上运行操作，可以使用 as_native 方法。

  let native_user: User = update_user_username.as_native().find_by_primary_key().execute(&session).await?;
  // action that requires native model
  authorize_user(&native_user);
  

  as_native 方法通过返回一个包含部分模型字段的新实例来工作。对于其他字段，它使用默认值。

• 建议的命名规范是 Purpose + Original Struct Name。例如： UpdateAdresssUser、UpdateDescriptionPost。

回调函数

回调是方便在模型在特定操作前后运行额外逻辑的方法。例如：

• 我们可以使用 before_insert 在插入前设置默认值和/或验证模型。
• 我们可以使用 after_update 来更新其他数据源，例如 Elasticsearch。

实现

1. 假设我们定义了一个自定义扩展，该扩展将在每次更新帖子时更新 Elasticsearch 文档。

   pub struct AppExtensions {
       pub elastic_client: ElasticClient,
   }
   

2. 现在我们可以实现一个利用此扩展的回调。

   #[charybdis_model(...)]
   pub struct Post {}
   
   impl ExtCallbacks for Post {
       type Extention = AppExtensions;
       type Error = AppError; // From<CharybdisError>
       
      // use before_insert to set default values
       async fn before_insert(
           &mut self,
           _session: &CachingSession,
           extension: &AppExtensions,
       ) -> Result<(), CustomError> {
           self.id = Uuid::new_v4();
           self.created_at = Utc::now();
           
           Ok(())
       }
       
       // use before_update to set updated_at
       async fn before_update(
           &mut self,
           _session: &CachingSession,
           extension: &AppExtensions,
       ) -> Result<(), CustomError> {
           self.updated_at = Utc::now();
           
           Ok(())
       }
   
       // use after_update to update elastic document
       async fn after_update(
           &mut self,
           _session: &CachingSession,
           extension: &AppExtensions,
       ) -> Result<(), CustomError> {
           extension.elastic_client.update(...).await?;
   
           Ok(())
       }
       
       // use after_delete to delete elastic document
       async fn after_delete(
           &mut self,
           _session: &CachingSession,
           extension: &AppExtensions,
       ) -> Result<(), CustomError> {
           extension.elastic_client.delete(...).await?;
   
           Ok(())
       }
   }
   

• 可能的回调

  • before_insert
  • before_update
  • before_delete
  • after_insert
  • after_update
  • after_delete

• 触发回调函数

  为了触发回调，我们使用 <operation>_cb 方法： insert_cb、update_cb、delete_cb 根据特性行为。这使得我们可以在 insert 和带有回调的插入（insert_cb）之间有明确的区分。就像在主操作中一样，我们可以在执行之前配置回调操作查询。

   use charybdis::operations::{DeleteWithCallbacks, InsertWithCallbacks, UpdateWithCallbacks};
  
   post.insert_cb(app_extensions).execute(&session).await;
   post.update_cb(app_extensions).execute(&session).await;
   post.delete_cb(app_extensions).consistency(Consistency::All).execute(&session).await;
  

集合

对于每个集合字段，我们得到以下内容：

• PUSH_<field_name>_QUERY 静态字符串
• PUSH_<field_name>_IF_EXISTS_QUERY 静态字符串
• PULL_<field_name>_QUERY 静态字符串
• PULL_<field_name>_IF_EXISTS_QUERY 静态字符串
• push_<field_name> 方法
• push_<field_name>_if_exists 方法
• pull_<field_name> 方法
• pull_<field_name>_if_exists 方法

1. 模型

   #[charybdis_model(
       table_name = users,
       partition_keys = [id],
       clustering_keys = []
   )]
   pub struct User {
       id: Uuid,
       tags: Set<Text>,
       post_ids: List<Uuid>,
       books_by_genre: Map<Text, Frozen<List<Text>>>,
   }
   

2. 生成的收集查询

   生成的查询将期望值作为第一个绑定值，而主键字段作为后续的绑定值。

   impl User {
       const PUSH_TAGS_QUERY: &'static str = "UPDATE users SET tags = tags + ? WHERE id = ?";
       const PUSH_TAGS_IF_EXISTS_QUERY: &'static str = "UPDATE users SET tags = tags + ? WHERE id = ? IF EXISTS";
   
       const PULL_TAGS_QUERY: &'static str = "UPDATE users SET tags = tags - ? WHERE id = ?";
       const PULL_TAGS_IF_EXISTS_QUERY: &'static str = "UPDATE users SET tags = tags - ? WHERE id = ? IF EXISTS";
        
       const PUSH_POST_IDS_QUERY: &'static str = "UPDATE users SET post_ids = post_ids + ? WHERE id = ?";
       const PUSH_POST_IDS_IF_EXISTS_QUERY: &'static str = "UPDATE users SET post_ids = post_ids + ? WHERE id = ? IF EXISTS";
   
       const PULL_POST_IDS_QUERY: &'static str = "UPDATE users SET post_ids = post_ids - ? WHERE id = ?";
       const PULL_POST_IDS_IF_EXISTS_QUERY: &'static str = "UPDATE users SET post_ids = post_ids - ? WHERE id = ? IF EXISTS";
   
       const PUSH_BOOKS_BY_GENRE_QUERY: &'static str = "UPDATE users SET books_by_genre = books_by_genre + ? WHERE id = ?";
       const PUSH_BOOKS_BY_GENRE_IF_EXISTS_QUERY: &'static str = "UPDATE users SET books_by_genre = books_by_genre + ? WHERE id = ? IF EXISTS";
       
       const PULL_BOOKS_BY_GENRE_QUERY: &'static str = "UPDATE users SET books_by_genre = books_by_genre - ? WHERE id = ?";
       const PULL_BOOKS_BY_GENRE_IF_EXISTS_QUERY: &'static str = "UPDATE users SET books_by_genre = books_by_genre - ? WHERE id = ? IF EXISTS";
   }
   

   现在我们可以在批量操作中使用此常量。

   let batch = User::batch();
   let users: Vec<User>;
   
   for user in users {
       batch.append_statement(User::PUSH_TAGS_QUERY, (vec![tag], user.id));
   }
   
   batch.execute(&session).await;
   

3. 生成的收集方法

   为每个集合字段生成了 push_to_<field_name> 和 pull_from_<field_name> 方法。

   let user: User::new();
   
   user.push_tags(tags: HashSet<T>).execute(&session).await;
   user.push_tags_if_exists(tags: HashSet<T>).execute(&session).await;
   
   user.pull_tags(tags: HashSet<T>).execute(&session).await;
   user.pull_tags_if_exists(tags: HashSet<T>).execute(&session).await;
   
   
   user.push_post_ids(ids: Vec<T>).execute(&session).await;
   user.push_post_ids_if_exists(ids: Vec<T>).execute(&session).await;
   
   user.pull_post_ids(ids: Vec<T>).execute(&session).await;
   user.pull_post_ids_if_exists(ids: Vec<T>).execute(&session).await;
   
   user.push_books_by_genre(map: HashMap<K, V>).execute(&session).await;
   user.push_books_by_genre_if_exists(map: HashMap<K, V>).execute(&session).await;
   
   user.pull_books_by_genre(map: HashMap<K, V>).execute(&session).await;
   user.pull_books_by_genre_if_exists(map: HashMap<K, V>).execute(&session).await;
   

忽略的字段

我们可以通过使用 #[charybdis(ignore)] 属性来忽略字段

#[charybdis_model(...)]
pub struct User {
    id: Uuid,
    #[charybdis(ignore)]
    organization: Option<Organization>,
}

因此，字段 organization 在所有操作中都将被忽略，并在从其他数据源反序列化时使用默认值。它可以用来存储不持久化在数据库中的数据。