Merge pull request #10021 from Turbo87/async-pagination

Implement async pagination

Author: Turbo87

src/controllers/helpers/pagination.rs

@@ -7,12 +7,14 @@ use crate::models::helpers::with_count::*;
 use crate::util::errors::{bad_request, AppResult};
 use crate::util::{HeaderMapExt, RequestUtils};
 
-use crate::util::diesel::prelude::*;
 use base64::{engine::general_purpose, Engine};
 use diesel::pg::Pg;
+use diesel::prelude::*;
 use diesel::query_builder::{AstPass, Query, QueryFragment, QueryId};
-use diesel::query_dsl::LoadQuery;
 use diesel::sql_types::BigInt;
+use diesel_async::AsyncPgConnection;
+use futures_util::future::BoxFuture;
+use futures_util::{FutureExt, TryStreamExt};
 use http::header;
 use indexmap::IndexMap;
 use serde::{Deserialize, Serialize};
@@ -250,16 +252,29 @@ pub(crate) struct PaginatedQuery<T> {
 }
 
 impl<T> PaginatedQuery<T> {
-    pub(crate) fn load<'a, U, Conn>(self, conn: &mut Conn) -> QueryResult<Paginated<U>>
+    pub fn load<'a, U>(
+        self,
+        conn: &'a mut AsyncPgConnection,
+    ) -> BoxFuture<'a, QueryResult<Paginated<U>>>
     where
-        Self: LoadQuery<'a, Conn, WithCount<U>>,
+        Self: diesel_async::methods::LoadQuery<'a, AsyncPgConnection, WithCount<U>>,
+        T: 'a,
+        U: Send + 'a,
     {
+        use diesel_async::methods::LoadQuery;
+
         let options = self.options.clone();
-        let records_and_total = self.internal_load(conn)?.collect::<QueryResult<_>>()?;
-        Ok(Paginated {
-            records_and_total,
-            options,
-        })
+        let future = self.internal_load(conn);
+
+        async move {
+            let records_and_total = future.await?.try_collect().await?;
+
+            Ok(Paginated {
+                records_and_total,
+                options,
+            })
+        }
+        .boxed()
     }
 }
 
@@ -272,8 +287,6 @@ impl<T: Query> Query for PaginatedQuery<T> {
     type SqlType = (T::SqlType, BigInt);
 }
 
-impl<T, DB> diesel::RunQueryDsl<DB> for PaginatedQuery<T> {}
-
 impl<T> QueryFragment<Pg> for PaginatedQuery<T>
 where
     T: QueryFragment<Pg>,
@@ -366,8 +379,6 @@ impl<
     type SqlType = (T::SqlType, BigInt);
 }
 
-impl<T, C, DB> diesel::RunQueryDsl<DB> for PaginatedQueryWithCountSubq<T, C> {}
-
 impl<T, C> QueryFragment<Pg> for PaginatedQueryWithCountSubq<T, C>
 where
     T: QueryFragment<Pg>,
@@ -390,16 +401,30 @@ where
 }
 
 impl<T, C> PaginatedQueryWithCountSubq<T, C> {
-    pub(crate) fn load<'a, U, Conn>(self, conn: &mut Conn) -> QueryResult<Paginated<U>>
+    pub fn load<'a, U>(
+        self,
+        conn: &'a mut AsyncPgConnection,
+    ) -> BoxFuture<'a, QueryResult<Paginated<U>>>
     where
-        Self: LoadQuery<'a, Conn, WithCount<U>>,
+        Self: diesel_async::methods::LoadQuery<'a, AsyncPgConnection, WithCount<U>> + Send,
+        C: 'a,
+        T: 'a,
+        U: Send + 'a,
     {
+        use diesel_async::methods::LoadQuery;
+
         let options = self.options.clone();
-        let records_and_total = self.internal_load(conn)?.collect::<QueryResult<_>>()?;
-        Ok(Paginated {
-            records_and_total,
-            options,
-        })
+        let future = self.internal_load(conn);
+
+        async move {
+            let records_and_total = future.await?.try_collect().await?;
+
+            Ok(Paginated {
+                records_and_total,
+                options,
+            })
+        }
+        .boxed()
     }
 }
 

src/controllers/keyword.rs

@@ -2,14 +2,12 @@ use crate::app::AppState;
 use crate::controllers::helpers::pagination::PaginationOptions;
 use crate::controllers::helpers::{pagination::Paginated, Paginate};
 use crate::models::Keyword;
-use crate::tasks::spawn_blocking;
 use crate::util::errors::AppResult;
 use crate::views::EncodableKeyword;
 use axum::extract::{Path, Query};
 use axum_extra::json;
 use axum_extra::response::ErasedJson;
 use diesel::prelude::*;
-use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
 use http::request::Parts;
 
 #[derive(Deserialize)]
@@ -30,23 +28,18 @@ pub async fn index(state: AppState, qp: Query<IndexQuery>, req: Parts) -> AppRes
 
     let query = query.pages_pagination(PaginationOptions::builder().gather(&req)?);
 
-    let conn = state.db_read().await?;
-    spawn_blocking(move || {
-        let conn: &mut AsyncConnectionWrapper<_> = &mut conn.into();
-
-        let data: Paginated<Keyword> = query.load(conn)?;
-        let total = data.total();
-        let kws = data
-            .into_iter()
-            .map(Keyword::into)
-            .collect::<Vec<EncodableKeyword>>();
-
-        Ok(json!({
-            "keywords": kws,
-            "meta": { "total": total },
-        }))
-    })
-    .await?
+    let mut conn = state.db_read().await?;
+    let data: Paginated<Keyword> = query.load(&mut conn).await?;
+    let total = data.total();
+    let kws = data
+        .into_iter()
+        .map(Keyword::into)
+        .collect::<Vec<EncodableKeyword>>();
+
+    Ok(json!({
+        "keywords": kws,
+        "meta": { "total": total },
+    }))
 }
 
 /// Handles the `GET /keywords/:keyword_id` route.