Skip to content

feat: use SqlDateTime abstraction from apalis-sql #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Himmelschmidt wants to merge 2 commits into
base: main
Choose a base branch
from
+125 −63
Open

feat: use SqlDateTime abstraction from apalis-sql #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c96d98b
feat: use SqlDateTime abstraction from apalis-sql
Himmelschmidt Dec 26, 2025
3a26ef9
chore: update apalis deps to git and add get_queue impl
Himmelschmidt Dec 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 82 additions & 30 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

17 changes: 9 additions & 8 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,21 +14,22 @@ categories = ["asynchronous", "database", "network-programming"]
publish = true

[features]
default = ["migrate", "tokio-comp"]
default = ["migrate", "tokio-comp", "chrono"]
migrate = ["sqlx/migrate", "sqlx/macros"]
async-std-comp = ["async-std", "sqlx/runtime-async-std-rustls"]
async-std-comp-native-tls = ["async-std", "sqlx/runtime-async-std-native-tls"]
tokio-comp = ["tokio", "sqlx/runtime-tokio-rustls"]
tokio-comp-native-tls = ["tokio", "sqlx/runtime-tokio-native-tls"]
chrono = ["apalis-sql/chrono", "sqlx/chrono"]
time = ["apalis-sql/time", "sqlx/time"]

[dependencies]
apalis-core = { version = "1.0.0-rc.1", default-features = false, features = [
apalis-core = { git = "https://github.com/apalis-dev/apalis", version = "1.0.0-rc.1", default-features = false, features = [
"sleep",
] }
apalis-sql = { version = "1.0.0-rc.1", default-features = false }
apalis-codec = { version = "0.1.0-rc.1", features = ["json"] }
apalis-sql = { git = "https://github.com/apalis-dev/apalis", version = "1.0.0-rc.1", default-features = false }
apalis-codec = { git = "https://github.com/apalis-dev/apalis", version = "0.1.0-rc.1", features = ["json"] }
serde = { version = "1", features = ["derive"], default-features = false }
chrono = { version = "0.4", features = ["serde"], default-features = false }
pin-project = "1.1.10"
serde_json = "1"
futures = "0.3.30"
Expand All @@ -44,11 +45,11 @@ ulid = { version = "1", features = ["serde"] }
[dependencies.sqlx]
version = "0.8.1"
default-features = false
features = ["chrono", "postgres", "json"]
features = ["postgres", "json"]

[dev-dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
once_cell = "1.19.0"
apalis = { version = "1.0.0-rc.1" }
apalis-workflow = { version = "0.1.0-rc.1" }
apalis = { git = "https://github.com/apalis-dev/apalis", version = "1.0.0-rc.1" }
apalis-workflow = { git = "https://github.com/apalis-dev/apalis", version = "0.1.0-rc.1" }
futures-util = "0.3.30"
15 changes: 8 additions & 7 deletions src/from_row.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use chrono::{DateTime, Utc};
use apalis_sql::{SqlDateTime, TaskRow};

#[derive(Debug)]
pub struct PgTaskRow {
pub job: Option<Vec<u8>>,
Expand All @@ -7,19 +8,19 @@ pub struct PgTaskRow {
pub status: Option<String>,
pub attempts: Option<i32>,
pub max_attempts: Option<i32>,
pub run_at: Option<DateTime<Utc>>,
pub run_at: Option<SqlDateTime>,
pub last_result: Option<serde_json::Value>,
pub lock_at: Option<DateTime<Utc>>,
pub lock_at: Option<SqlDateTime>,
pub lock_by: Option<String>,
pub done_at: Option<DateTime<Utc>>,
pub done_at: Option<SqlDateTime>,
pub priority: Option<i32>,
pub metadata: Option<serde_json::Value>,
}
impl TryInto<apalis_sql::from_row::TaskRow> for PgTaskRow {
impl TryInto<TaskRow> for PgTaskRow {
type Error = sqlx::Error;

fn try_into(self) -> Result<apalis_sql::from_row::TaskRow, Self::Error> {
Ok(apalis_sql::from_row::TaskRow {
fn try_into(self) -> Result<TaskRow, Self::Error> {
Ok(TaskRow {
job: self.job.unwrap_or_default(),
id: self
.id
Expand Down
7 changes: 4 additions & 3 deletions src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ mod ack;
mod fetcher;
mod from_row;

pub type PgContext = apalis_sql::context::SqlContext<PgPool>;
pub type PgContext = apalis_sql::context::SqlContext<Postgres>;
mod queries;
pub mod shared;
pub mod sink;
Expand Down Expand Up @@ -252,8 +252,9 @@ where
type CompactStream = TaskStream<PgTask<CompactType>, Self::Error>;

fn get_queue(&self) -> Queue {
self.config.queue().clone()
Queue::from(self.config.queue().to_string())
}

fn poll_compact(self, worker: &WorkerContext) -> Self::CompactStream {
self.poll_basic(worker).boxed()
}
Expand Down Expand Up @@ -349,7 +350,7 @@ where
type CompactStream = TaskStream<PgTask<CompactType>, Self::Error>;

fn get_queue(&self) -> Queue {
self.config.queue().clone()
Queue::from(self.config.queue().to_string())
}

fn poll_compact(self, worker: &WorkerContext) -> Self::CompactStream {
Expand Down
Loading