feat(server): start db query abstraction

server/src/api/mod.rs

@@ -4,16 +4,11 @@ use axum::extract::{Path, Query, State};
 use axum::routing::get;
 use axum::Json;
 use axum::Router;
-use sea_orm::entity::EntityTrait;
-use sea_orm::query::QueryOrder;
 use sea_orm::ModelTrait;
-use sea_orm::PaginatorTrait;
 
 use pagination::PaginatedResponse;
 
-use crate::db::entities::package;
+use crate::db;
-use crate::db::entities::package_license;
-use crate::db::entities::repo;
 
 pub fn router() -> Router<crate::Global> {
     Router::new()
@@ -26,22 +21,24 @@ pub fn router() -> Router<crate::Global> {
 async fn get_repos(
     State(global): State<crate::Global>,
     Query(pagination): Query<pagination::Query>,
-) -> crate::Result<Json<PaginatedResponse<repo::Model>>> {
+) -> crate::Result<Json<PaginatedResponse<db::repo::Model>>> {
-    let repos = repo::Entity::find()
+    let (total_pages, repos) = global
-        .order_by_asc(repo::Column::Id)
+        .db
-        .paginate(&global.db, pagination.per_page.unwrap_or(25))
+        .repos(
-        .fetch_page(pagination.page.unwrap_or(1) - 1)
+            pagination.per_page.unwrap_or(25),
+            pagination.page.unwrap_or(1) - 1,
+        )
         .await?;
-
+    Ok(Json(pagination.res(total_pages, repos)))
-    Ok(Json(pagination.res(repos)))
 }
 
 async fn get_single_repo(
     State(global): State<crate::Global>,
     Path(id): Path<i32>,
-) -> crate::Result<Json<repo::Model>> {
+) -> crate::Result<Json<db::repo::Model>> {
-    let repo = repo::Entity::find_by_id(id)
+    let repo = global
-        .one(&global.db)
+        .db
+        .repo(id)
         .await?
         .ok_or(axum::http::StatusCode::NOT_FOUND)?;
 
@@ -51,20 +48,22 @@ async fn get_single_repo(
 async fn get_packages(
     State(global): State<crate::Global>,
     Query(pagination): Query<pagination::Query>,
-) -> crate::Result<Json<PaginatedResponse<package::Model>>> {
+) -> crate::Result<Json<PaginatedResponse<db::package::Model>>> {
-    let pkgs = package::Entity::find()
+    let (total_pages, pkgs) = global
-        .order_by_asc(package::Column::Id)
+        .db
-        .paginate(&global.db, pagination.per_page.unwrap_or(25))
+        .packages(
-        .fetch_page(pagination.page.unwrap_or(1) - 1)
+            pagination.per_page.unwrap_or(25),
+            pagination.page.unwrap_or(1) - 1,
+        )
         .await?;
 
-    Ok(Json(pagination.res(pkgs)))
+    Ok(Json(pagination.res(total_pages, pkgs)))
 }
 
 #[derive(serde::Serialize)]
 pub struct PackageRes {
     #[serde(flatten)]
-    entry: package::Model,
+    entry: db::package::Model,
     licenses: Vec<String>,
 }
 
@@ -72,12 +71,13 @@ async fn get_single_package(
     State(global): State<crate::Global>,
     Path(id): Path<i32>,
 ) -> crate::Result<Json<PackageRes>> {
-    let entry = package::Entity::find_by_id(id)
+    let entry = global
-        .one(&global.db)
+        .db
+        .package(id)
         .await?
         .ok_or(axum::http::StatusCode::NOT_FOUND)?;
     let licenses = entry
-        .find_related(package_license::Entity)
+        .find_related(db::PackageLicense)
         .all(&global.db)
         .await?
         .iter()

server/src/api/pagination.rs

@@ -16,15 +16,21 @@ where
 {
     pub page: u64,
     pub per_page: u64,
+    pub total_pages: u64,
     pub count: usize,
     pub items: Vec<T>,
 }
 
 impl Query {
-    pub fn res<T: for<'de> Serialize>(self, items: Vec<T>) -> PaginatedResponse<T> {
+    pub fn res<T: for<'de> Serialize>(
+        self,
+        total_pages: u64,
+        items: Vec<T>,
+    ) -> PaginatedResponse<T> {
         PaginatedResponse {
             page: self.page.unwrap_or(DEFAULT_PAGE),
             per_page: self.per_page.unwrap_or(DEFAULT_PER_PAGE),
+            total_pages,
             count: items.len(),
             items,
         }

server/src/cli.rs

@@ -65,7 +65,7 @@ impl Cli {
 
         debug!("Connecting to database with URL {}", db_url);
 
-        let db = crate::db::init(db_url).await?;
+        let db = crate::db::RieterDb::connect(db_url).await?;
         // let db = crate::db::init("postgres://rieter:rieter@localhost:5432/rieter")
         // .await
         // .unwrap();

server/src/db/conn.rs

@@ -0,0 +1,61 @@
+use super::RieterDb;
+use sea_orm::{DbBackend, DbErr, ExecResult, QueryResult, Statement};
+use std::{future::Future, pin::Pin};
+
+// Allows RieterDb objects to be passed to ORM functions
+impl sea_orm::ConnectionTrait for RieterDb {
+    fn get_database_backend(&self) -> DbBackend {
+        self.conn.get_database_backend()
+    }
+    fn execute<'life0, 'async_trait>(
+        &'life0 self,
+        stmt: Statement,
+    ) -> Pin<Box<dyn Future<Output = std::result::Result<ExecResult, DbErr>> + Send + 'async_trait>>
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+    {
+        self.conn.execute(stmt)
+    }
+    fn execute_unprepared<'life0, 'life1, 'async_trait>(
+        &'life0 self,
+        sql: &'life1 str,
+    ) -> Pin<Box<dyn Future<Output = std::result::Result<ExecResult, DbErr>> + Send + 'async_trait>>
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+        'life1: 'async_trait,
+    {
+        self.conn.execute_unprepared(sql)
+    }
+    fn query_one<'life0, 'async_trait>(
+        &'life0 self,
+        stmt: Statement,
+    ) -> Pin<
+        Box<
+            dyn Future<Output = std::result::Result<Option<QueryResult>, DbErr>>
+                + Send
+                + 'async_trait,
+        >,
+    >
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+    {
+        self.conn.query_one(stmt)
+    }
+    fn query_all<'life0, 'async_trait>(
+        &'life0 self,
+        stmt: Statement,
+    ) -> Pin<
+        Box<
+            dyn Future<Output = std::result::Result<Vec<QueryResult>, DbErr>> + Send + 'async_trait,
+        >,
+    >
+    where
+        Self: 'async_trait,
+        'life0: 'async_trait,
+    {
+        self.conn.query_all(stmt)
+    }
+}

server/src/db/mod.rs

@@ -1,17 +1,69 @@
+mod conn;
 pub mod entities;
 mod migrator;
 
 use migrator::Migrator;
+use sea_orm::ColumnTrait;
 use sea_orm::ConnectOptions;
 use sea_orm::Database;
+use sea_orm::DatabaseConnection;
+use sea_orm::EntityTrait;
+use sea_orm::PaginatorTrait;
+use sea_orm::QueryFilter;
+use sea_orm::QueryOrder;
 use sea_orm_migration::MigratorTrait;
 
-pub async fn init<C: Into<ConnectOptions>>(
+pub use entities::prelude::*;
-    opt: C,
+pub use entities::*;
-) -> Result<sea_orm::DatabaseConnection, sea_orm::DbErr> {
+
+type Result<T> = std::result::Result<T, sea_orm::DbErr>;
+
+#[derive(Clone, Debug)]
+pub struct RieterDb {
+    pub conn: DatabaseConnection,
+}
+
+impl RieterDb {
+    pub async fn connect<C: Into<ConnectOptions>>(opt: C) -> Result<Self> {
         let db = Database::connect(opt).await?;
 
         Migrator::up(&db, None).await?;
 
-    Ok(db)
+        Ok(Self { conn: db })
+    }
+
+    pub async fn repos(&self, per_page: u64, page: u64) -> Result<(u64, Vec<repo::Model>)> {
+        let paginator = Repo::find()
+            .order_by_asc(repo::Column::Id)
+            .paginate(&self.conn, per_page);
+        let repos = paginator.fetch_page(page).await?;
+        let total_pages = paginator.num_pages().await?;
+
+        Ok((total_pages, repos))
+    }
+
+    pub async fn repo(&self, id: i32) -> Result<Option<repo::Model>> {
+        repo::Entity::find_by_id(id).one(&self.conn).await
+    }
+
+    pub async fn repo_by_name(&self, name: &str) -> Result<Option<repo::Model>> {
+        Repo::find()
+            .filter(repo::Column::Name.eq(name))
+            .one(&self.conn)
+            .await
+    }
+
+    pub async fn packages(&self, per_page: u64, page: u64) -> Result<(u64, Vec<package::Model>)> {
+        let paginator = Package::find()
+            .order_by_asc(package::Column::Id)
+            .paginate(&self.conn, per_page);
+        let packages = paginator.fetch_page(page).await?;
+        let total_pages = paginator.num_pages().await?;
+
+        Ok((total_pages, packages))
+    }
+
+    pub async fn package(&self, id: i32) -> Result<Option<package::Model>> {
+        package::Entity::find_by_id(id).one(&self.conn).await
+    }
 }

server/src/main.rs

@@ -7,7 +7,6 @@ mod repo;
 use clap::Parser;
 pub use error::{Result, ServerError};
 use repo::RepoGroupManager;
-use sea_orm::DatabaseConnection;
 use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
 
@@ -23,7 +22,7 @@ pub struct Config {
 pub struct Global {
     config: Config,
     repo_manager: Arc<RwLock<RepoGroupManager>>,
-    db: DatabaseConnection,
+    db: db::RieterDb,
 }
 
 #[tokio::main]

server/src/repo/mod.rs

@@ -5,9 +5,7 @@ pub use manager::RepoGroupManager;
 
 use std::path::PathBuf;
 
-use crate::db::entities::{
+use crate::db;
-    package as db_package, package_license as db_package_license, repo as db_repo,
-};
 use axum::body::Body;
 use axum::extract::{BodyStream, Path, State};
 use axum::http::Request;
@@ -129,30 +127,27 @@ async fn post_package_archive(
             tracing::info!("Added '{}' to repository '{}'", pkg.file_name(), repo);
 
             // Query the repo for its ID, or create it if it does not already exist
-            let res = db_repo::Entity::find()
+            let res = global.db.repo_by_name(&repo).await?;
-                .filter(db_repo::Column::Name.eq(&repo))
-                .one(&global.db)
-                .await?;
 
             let repo_id = if let Some(repo_entity) = res {
                 repo_entity.id
             } else {
-                let model = db_repo::ActiveModel {
+                let model = db::repo::ActiveModel {
                     name: sea_orm::Set(repo.clone()),
                     ..Default::default()
                 };
 
-                db_repo::Entity::insert(model)
+                db::Repo::insert(model)
                     .exec(&global.db)
                     .await?
                     .last_insert_id
             };
 
             // If the package already exists in the database, we remove it first
-            let res = db_package::Entity::find()
+            let res = db::Package::find()
-                .filter(db_package::Column::RepoId.eq(repo_id))
+                .filter(db::package::Column::RepoId.eq(repo_id))
-                .filter(db_package::Column::Name.eq(&pkg.info.name))
+                .filter(db::package::Column::Name.eq(&pkg.info.name))
-                .filter(db_package::Column::Arch.eq(&pkg.info.arch))
+                .filter(db::package::Column::Arch.eq(&pkg.info.arch))
                 .one(&global.db)
                 .await?;
 
@@ -161,12 +156,12 @@ async fn post_package_archive(
             }
 
             // Insert the package's data into the database
-            let mut model: db_package::ActiveModel = pkg.clone().into();
+            let mut model: db::package::ActiveModel = pkg.clone().into();
             model.repo_id = sea_orm::Set(repo_id);
 
             let pkg_entry = model.insert(&global.db).await?;
-            db_package_license::Entity::insert_many(pkg.info.licenses.iter().map(|s| {
+            db::PackageLicense::insert_many(pkg.info.licenses.iter().map(|s| {
-                db_package_license::ActiveModel {
+                db::package_license::ActiveModel {
                     package_id: sea_orm::Set(pkg_entry.id),
                     value: sea_orm::Set(s.to_string()),
                 }
@@ -197,10 +192,7 @@ async fn delete_repo(
             .await??;
 
     if repo_removed {
-        let res = db_repo::Entity::find()
+        let res = global.db.repo_by_name(&repo).await?;
-            .filter(db_repo::Column::Name.eq(&repo))
-            .one(&global.db)
-            .await?;
 
         if let Some(repo_entry) = res {
             repo_entry.delete(&global.db).await?;
@@ -231,16 +223,13 @@ async fn delete_arch_repo(
     .await??;
 
     if repo_removed {
-        let res = db_repo::Entity::find()
+        let res = global.db.repo_by_name(&repo).await?;
-            .filter(db_repo::Column::Name.eq(&repo))
-            .one(&global.db)
-            .await?;
 
         if let Some(repo_entry) = res {
             // Also remove all packages for that architecture from database
-            db_package::Entity::delete_many()
+            db::Package::delete_many()
-                .filter(db_package::Column::RepoId.eq(repo_entry.id))
+                .filter(db::package::Column::RepoId.eq(repo_entry.id))
-                .filter(db_package::Column::Arch.eq(&arch))
+                .filter(db::package::Column::Arch.eq(&arch))
                 .exec(&global.db)
                 .await?;
         }
@@ -265,18 +254,15 @@ async fn delete_package(
     .await??;
 
     if let Some((name, version, release, arch)) = res {
-        let res = db_repo::Entity::find()
+        let res = global.db.repo_by_name(&repo).await?;
-            .filter(db_repo::Column::Name.eq(&repo))
-            .one(&global.db)
-            .await?;
 
         if let Some(repo_entry) = res {
             // Also remove entry from database
-            let res = db_package::Entity::find()
+            let res = db::Package::find()
-                .filter(db_package::Column::RepoId.eq(repo_entry.id))
+                .filter(db::package::Column::RepoId.eq(repo_entry.id))
-                .filter(db_package::Column::Name.eq(name))
+                .filter(db::package::Column::Name.eq(name))
-                .filter(db_package::Column::Version.eq(format!("{}-{}", version, release)))
+                .filter(db::package::Column::Version.eq(format!("{}-{}", version, release)))
-                .filter(db_package::Column::Arch.eq(arch))
+                .filter(db::package::Column::Arch.eq(arch))
                 .one(&global.db)
                 .await?;