#![allow(clippy::cast_possible_truncation)]
#![allow(clippy::cast_sign_loss)]
#![allow(clippy::cast_lossless)]
#![allow(clippy::result_large_err)] #![allow(clippy::borrow_as_ptr)] #![allow(clippy::if_not_else)] #![allow(clippy::implicit_clone)] #![allow(clippy::map_unwrap_or)] #![allow(clippy::redundant_closure)]
use crate::ffi;
use crate::types;
use sqlmodel_core::{
Connection, Cx, Error, IsolationLevel, Outcome, PreparedStatement, Row, TransactionOps, Value,
error::{ConfigError, ConnectionError, ConnectionErrorKind, QueryError, QueryErrorKind},
row::ColumnInfo,
};
use std::ffi::{CStr, CString, c_int};
use std::future::Future;
use std::ptr;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SqliteErrorCode {
primary: c_int,
extended: c_int,
}
impl SqliteErrorCode {
const fn from_result_codes(result: c_int, extended: c_int) -> Self {
Self {
primary: result & 0xff,
extended,
}
}
#[must_use]
pub const fn primary(self) -> c_int {
self.primary
}
#[must_use]
pub const fn extended(self) -> c_int {
self.extended
}
}
impl std::fmt::Display for SqliteErrorCode {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
formatter,
"SQLite result code {} (extended {})",
self.primary, self.extended
)
}
}
impl std::error::Error for SqliteErrorCode {}
#[must_use]
pub fn sqlite_error_code(error: &Error) -> Option<SqliteErrorCode> {
std::error::Error::source(error)?
.downcast_ref::<SqliteErrorCode>()
.copied()
}
#[cfg(feature = "console")]
use sqlmodel_console::{ConsoleAware, SqlModelConsole};
#[derive(Debug, Clone)]
pub struct SqliteConfig {
pub path: String,
pub flags: OpenFlags,
pub busy_timeout_ms: u32,
}
#[derive(Debug, Clone, Copy, Default)]
pub struct OpenFlags {
pub read_only: bool,
pub read_write: bool,
pub create: bool,
pub uri: bool,
pub no_mutex: bool,
pub full_mutex: bool,
pub shared_cache: bool,
pub private_cache: bool,
}
impl OpenFlags {
pub fn read_only() -> Self {
Self {
read_only: true,
..Default::default()
}
}
pub fn read_write() -> Self {
Self {
read_write: true,
..Default::default()
}
}
pub fn create_read_write() -> Self {
Self {
read_write: true,
create: true,
..Default::default()
}
}
fn to_sqlite_flags(self) -> c_int {
let mut flags = 0;
if self.read_only {
flags |= ffi::SQLITE_OPEN_READONLY;
}
if self.read_write {
flags |= ffi::SQLITE_OPEN_READWRITE;
}
if self.create {
flags |= ffi::SQLITE_OPEN_CREATE;
}
if self.uri {
flags |= ffi::SQLITE_OPEN_URI;
}
if self.no_mutex {
flags |= ffi::SQLITE_OPEN_NOMUTEX;
}
if self.full_mutex {
flags |= ffi::SQLITE_OPEN_FULLMUTEX;
}
if self.shared_cache {
flags |= ffi::SQLITE_OPEN_SHAREDCACHE;
} else {
flags |= ffi::SQLITE_OPEN_PRIVATECACHE;
}
if flags & (ffi::SQLITE_OPEN_READONLY | ffi::SQLITE_OPEN_READWRITE) == 0 {
flags |= ffi::SQLITE_OPEN_READWRITE | ffi::SQLITE_OPEN_CREATE;
}
flags
}
}
impl Default for SqliteConfig {
fn default() -> Self {
Self {
path: ":memory:".to_string(),
flags: OpenFlags::create_read_write(),
busy_timeout_ms: 5000,
}
}
}
impl SqliteConfig {
pub fn file(path: impl Into<String>) -> Self {
Self {
path: path.into(),
flags: OpenFlags::create_read_write(),
busy_timeout_ms: 5000,
}
}
pub fn memory() -> Self {
Self::default()
}
pub fn flags(mut self, flags: OpenFlags) -> Self {
self.flags = flags;
self
}
pub fn busy_timeout(mut self, ms: u32) -> Self {
self.busy_timeout_ms = ms;
self
}
}
struct SqliteInner {
db: *mut ffi::sqlite3,
in_transaction: bool,
}
unsafe impl Send for SqliteInner {}
pub struct SqliteConnection {
inner: Mutex<SqliteInner>,
path: String,
uses_shared_cache: bool,
#[cfg(feature = "console")]
console: Option<Arc<SqlModelConsole>>,
}
unsafe impl Send for SqliteConnection {}
unsafe impl Sync for SqliteConnection {}
impl SqliteConnection {
pub fn open(config: &SqliteConfig) -> Result<Self, Error> {
if config.flags.shared_cache && config.flags.private_cache {
return Err(Error::Config(ConfigError {
message: "SQLite shared_cache and private_cache flags are mutually exclusive"
.to_string(),
source: None,
}));
}
let busy_timeout_ms = c_int::try_from(config.busy_timeout_ms).map_err(|_| {
Error::Config(ConfigError {
message: format!(
"SQLite busy timeout {}ms exceeds the native {}ms limit",
config.busy_timeout_ms,
c_int::MAX
),
source: None,
})
})?;
let c_path = CString::new(config.path.as_str()).map_err(|_| {
Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: "Invalid path: contains null byte".to_string(),
source: None,
})
})?;
let mut db: *mut ffi::sqlite3 = ptr::null_mut();
let flags = config.flags.to_sqlite_flags();
let rc = unsafe { ffi::sqlite3_open_v2(c_path.as_ptr(), &mut db, flags, ptr::null()) };
if rc != ffi::SQLITE_OK {
let error_code = sqlite_error_code_from_db(db, rc);
let msg = if !db.is_null() {
unsafe {
let err_ptr = ffi::sqlite3_errmsg(db);
let msg = CStr::from_ptr(err_ptr).to_string_lossy().into_owned();
ffi::sqlite3_close(db);
msg
}
} else {
ffi::error_string(rc).to_string()
};
return Err(Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: format!("Failed to open database: {}", msg),
source: Some(Box::new(error_code)),
}));
}
if busy_timeout_ms > 0 {
let busy_rc = unsafe { ffi::sqlite3_busy_timeout(db, busy_timeout_ms) };
if busy_rc != ffi::SQLITE_OK {
let error_code = sqlite_error_code_from_db(db, busy_rc);
let msg = unsafe { CStr::from_ptr(ffi::sqlite3_errmsg(db)) }
.to_string_lossy()
.into_owned();
unsafe { ffi::sqlite3_close(db) };
return Err(Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: format!("Failed to configure SQLite busy timeout: {msg}"),
source: Some(Box::new(error_code)),
}));
}
}
Ok(Self {
inner: Mutex::new(SqliteInner {
db,
in_transaction: false,
}),
path: config.path.clone(),
uses_shared_cache: connection_uses_shared_cache(config),
#[cfg(feature = "console")]
console: None,
})
}
pub fn open_memory() -> Result<Self, Error> {
Self::open(&SqliteConfig::memory())
}
pub fn open_file(path: impl Into<String>) -> Result<Self, Error> {
Self::open(&SqliteConfig::file(path))
}
pub fn path(&self) -> &str {
&self.path
}
pub fn execute_raw(&self, sql: &str) -> Result<(), Error> {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
let c_sql = CString::new(sql).map_err(|_| {
Error::Query(QueryError {
kind: QueryErrorKind::Syntax,
sql: Some(sql.to_string()),
sqlstate: None,
message: "SQL contains null byte".to_string(),
detail: None,
hint: None,
position: None,
source: None,
})
})?;
let mut errmsg: *mut std::ffi::c_char = ptr::null_mut();
let rc = unsafe {
ffi::sqlite3_exec(inner.db, c_sql.as_ptr(), None, ptr::null_mut(), &mut errmsg)
};
if rc != ffi::SQLITE_OK {
let error_code = sqlite_error_code_from_db(inner.db, rc);
let msg = if !errmsg.is_null() {
let msg = unsafe { CStr::from_ptr(errmsg).to_string_lossy().into_owned() };
unsafe { ffi::sqlite3_free(errmsg.cast()) };
msg
} else {
ffi::error_string(rc).to_string()
};
return Err(Error::Query(QueryError {
kind: error_code_to_kind(rc),
sql: Some(sql.to_string()),
sqlstate: None,
message: msg,
detail: None,
hint: None,
position: None,
source: Some(Box::new(error_code)),
}));
}
Ok(())
}
pub fn backup_to_path(&self, dest_path: impl AsRef<str>) -> Result<(), Error> {
let dest = SqliteConnection::open(
&SqliteConfig::file(dest_path.as_ref()).flags(OpenFlags::create_read_write()),
)?;
self.backup_to_connection(&dest)
}
pub fn backup_to_connection(&self, dest: &SqliteConnection) -> Result<(), Error> {
if std::ptr::eq(self, dest) {
return Err(Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: "SQLite backup source and destination must be different connections"
.to_string(),
source: None,
}));
}
if dest.uses_shared_cache {
return Err(Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: "SQLite backup destinations cannot use shared-cache mode".to_string(),
source: None,
}));
}
let self_first = (std::ptr::from_ref(self) as usize) <= (std::ptr::from_ref(dest) as usize);
let (source_guard, dest_guard) = if self_first {
let source_guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
let dest_guard = dest.inner.lock().unwrap_or_else(|e| e.into_inner());
(source_guard, dest_guard)
} else {
let dest_guard = dest.inner.lock().unwrap_or_else(|e| e.into_inner());
let source_guard = self.inner.lock().unwrap_or_else(|e| e.into_inner());
(source_guard, dest_guard)
};
let source_db = source_guard.db;
let dest_db = dest_guard.db;
let source_busy_timeout_ms = sqlite_busy_timeout_ms(source_db)?;
let dest_busy_timeout_ms = sqlite_busy_timeout_ms(dest_db)?;
let main = CString::new("main").expect("static sqlite db name");
let backup =
unsafe { ffi::sqlite3_backup_init(dest_db, main.as_ptr(), source_db, main.as_ptr()) };
if backup.is_null() {
let result_code = unsafe { ffi::sqlite3_errcode(dest_db) };
let error_code = sqlite_error_code_from_db(dest_db, result_code);
let msg = unsafe { CStr::from_ptr(ffi::sqlite3_errmsg(dest_db)) }
.to_string_lossy()
.into_owned();
return Err(Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: format!("SQLite backup init failed: {msg}"),
source: Some(Box::new(error_code)),
}));
}
let busy_deadline = Instant::now()
+ Duration::from_millis(
u64::try_from(source_busy_timeout_ms.max(dest_busy_timeout_ms))
.expect("SQLite busy timeout is non-negative"),
);
let _busy_timeout_guard = BackupBusyTimeoutGuard::disable(
source_db,
dest_db,
source_busy_timeout_ms,
dest_busy_timeout_ms,
);
let mut rc = unsafe { ffi::sqlite3_backup_step(backup, 100) };
loop {
if rc == ffi::SQLITE_DONE {
break;
}
if rc == ffi::SQLITE_OK {
rc = unsafe { ffi::sqlite3_backup_step(backup, 100) };
continue;
}
if rc == ffi::SQLITE_BUSY || rc == ffi::SQLITE_LOCKED {
let now = Instant::now();
if now >= busy_deadline {
break;
}
std::thread::sleep(
Duration::from_millis(50).min(busy_deadline.saturating_duration_since(now)),
);
if Instant::now() >= busy_deadline {
break;
}
rc = unsafe { ffi::sqlite3_backup_step(backup, 100) };
continue;
}
break;
}
let backup_error = if rc != ffi::SQLITE_DONE && rc != ffi::SQLITE_OK {
Some(backup_step_error(dest_db, rc))
} else {
None
};
let finish_rc = unsafe { ffi::sqlite3_backup_finish(backup) };
if let Some(error) = backup_error {
return Err(error);
}
if finish_rc != ffi::SQLITE_OK {
let error_code = sqlite_error_code_from_db(dest_db, finish_rc);
let msg = unsafe { CStr::from_ptr(ffi::sqlite3_errmsg(dest_db)) }
.to_string_lossy()
.into_owned();
return Err(Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: format!(
"SQLite backup finish failed: {} ({})",
msg,
ffi::error_string(finish_rc)
),
source: Some(Box::new(error_code)),
}));
}
Ok(())
}
pub fn last_insert_rowid(&self) -> i64 {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
unsafe { ffi::sqlite3_last_insert_rowid(inner.db) }
}
pub fn changes(&self) -> i32 {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
unsafe { ffi::sqlite3_changes(inner.db) }
}
pub fn query_sync(&self, sql: &str, params: &[Value]) -> Result<Vec<Row>, Error> {
#[cfg(feature = "console")]
let start = std::time::Instant::now();
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
let stmt = prepare_stmt(inner.db, sql)?;
for (i, param) in params.iter().enumerate() {
let rc = unsafe { types::bind_value(stmt, (i + 1) as c_int, param) };
if rc != ffi::SQLITE_OK {
let error = bind_error(inner.db, sql, i + 1, rc);
unsafe { ffi::sqlite3_finalize(stmt) };
return Err(error);
}
}
let col_count = unsafe { ffi::sqlite3_column_count(stmt) };
let mut col_names = Vec::with_capacity(col_count as usize);
for i in 0..col_count {
let name =
unsafe { types::column_name(stmt, i) }.unwrap_or_else(|| format!("col{}", i));
col_names.push(name);
}
let columns = Arc::new(ColumnInfo::new(col_names.clone()));
let mut rows = Vec::new();
loop {
let rc = unsafe { ffi::sqlite3_step(stmt) };
match rc {
ffi::SQLITE_ROW => {
let mut values = Vec::with_capacity(col_count as usize);
for i in 0..col_count {
let value = unsafe { types::read_column(stmt, i) };
values.push(value);
}
rows.push(Row::with_columns(Arc::clone(&columns), values));
}
ffi::SQLITE_DONE => break,
_ => {
let error = step_error(inner.db, sql, rc);
unsafe { ffi::sqlite3_finalize(stmt) };
return Err(error);
}
}
}
unsafe { ffi::sqlite3_finalize(stmt) };
#[cfg(feature = "console")]
{
let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
self.emit_query_result(sql, &col_names, &rows, elapsed_ms);
}
Ok(rows)
}
pub fn execute_sync(&self, sql: &str, params: &[Value]) -> Result<u64, Error> {
#[cfg(feature = "console")]
let start = std::time::Instant::now();
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
let stmt = prepare_stmt(inner.db, sql)?;
for (i, param) in params.iter().enumerate() {
let rc = unsafe { types::bind_value(stmt, (i + 1) as c_int, param) };
if rc != ffi::SQLITE_OK {
let error = bind_error(inner.db, sql, i + 1, rc);
unsafe { ffi::sqlite3_finalize(stmt) };
return Err(error);
}
}
let execution_error = loop {
let rc = unsafe { ffi::sqlite3_step(stmt) };
match rc {
ffi::SQLITE_ROW => continue,
ffi::SQLITE_DONE => break None,
_ => break Some(step_error(inner.db, sql, rc)),
}
};
unsafe { ffi::sqlite3_finalize(stmt) };
if let Some(error) = execution_error {
return Err(error);
}
let changes = unsafe { ffi::sqlite3_changes(inner.db) };
#[cfg(feature = "console")]
{
let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
self.emit_execute_timing(sql, changes as u64, elapsed_ms);
}
Ok(changes as u64)
}
fn insert_sync(&self, sql: &str, params: &[Value]) -> Result<i64, Error> {
self.execute_sync(sql, params)?;
Ok(self.last_insert_rowid())
}
fn begin_sync(&self, isolation: IsolationLevel) -> Result<(), Error> {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
if inner.in_transaction {
return Err(Error::Query(QueryError {
kind: QueryErrorKind::Database,
sql: None,
sqlstate: None,
message: "Already in a transaction".to_string(),
detail: None,
hint: None,
position: None,
source: None,
}));
}
let begin_sql = match isolation {
IsolationLevel::Serializable => "BEGIN EXCLUSIVE",
IsolationLevel::RepeatableRead | IsolationLevel::ReadCommitted => "BEGIN IMMEDIATE",
IsolationLevel::ReadUncommitted => "BEGIN DEFERRED",
};
drop(inner); self.execute_raw(begin_sql)?;
let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.in_transaction = true;
self.emit_transaction_state("BEGIN");
Ok(())
}
fn commit_sync(&self) -> Result<(), Error> {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
if !inner.in_transaction {
return Err(Error::Query(QueryError {
kind: QueryErrorKind::Database,
sql: None,
sqlstate: None,
message: "Not in a transaction".to_string(),
detail: None,
hint: None,
position: None,
source: None,
}));
}
drop(inner);
self.execute_raw("COMMIT")?;
let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.in_transaction = false;
self.emit_transaction_state("COMMIT");
Ok(())
}
fn rollback_sync(&self) -> Result<(), Error> {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
if !inner.in_transaction {
return Err(Error::Query(QueryError {
kind: QueryErrorKind::Database,
sql: None,
sqlstate: None,
message: "Not in a transaction".to_string(),
detail: None,
hint: None,
position: None,
source: None,
}));
}
drop(inner);
self.execute_raw("ROLLBACK")?;
let mut inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
inner.in_transaction = false;
self.emit_transaction_state("ROLLBACK");
Ok(())
}
}
impl Drop for SqliteConnection {
fn drop(&mut self) {
if let Ok(inner) = self.inner.lock() {
if !inner.db.is_null() {
unsafe {
ffi::sqlite3_close_v2(inner.db);
}
}
}
}
}
pub struct SqliteTransaction<'conn> {
conn: &'conn SqliteConnection,
committed: bool,
}
impl<'conn> SqliteTransaction<'conn> {
fn new(conn: &'conn SqliteConnection) -> Self {
Self {
conn,
committed: false,
}
}
}
impl Drop for SqliteTransaction<'_> {
fn drop(&mut self) {
if !self.committed {
let _ = self.conn.rollback_sync();
}
}
}
impl Connection for SqliteConnection {
type Tx<'conn>
= SqliteTransaction<'conn>
where
Self: 'conn;
fn dialect(&self) -> sqlmodel_core::Dialect {
sqlmodel_core::Dialect::Sqlite
}
fn query(
&self,
_cx: &Cx,
sql: &str,
params: &[Value],
) -> impl Future<Output = Outcome<Vec<Row>, Error>> + Send {
let result = self.query_sync(sql, params);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn query_one(
&self,
_cx: &Cx,
sql: &str,
params: &[Value],
) -> impl Future<Output = Outcome<Option<Row>, Error>> + Send {
let result = self.query_sync(sql, params).map(|mut rows| rows.pop());
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn execute(
&self,
_cx: &Cx,
sql: &str,
params: &[Value],
) -> impl Future<Output = Outcome<u64, Error>> + Send {
let result = self.execute_sync(sql, params);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn insert(
&self,
_cx: &Cx,
sql: &str,
params: &[Value],
) -> impl Future<Output = Outcome<i64, Error>> + Send {
let result = self.insert_sync(sql, params);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn batch(
&self,
_cx: &Cx,
statements: &[(String, Vec<Value>)],
) -> impl Future<Output = Outcome<Vec<u64>, Error>> + Send {
let mut results = Vec::with_capacity(statements.len());
let mut error = None;
for (sql, params) in statements {
match self.execute_sync(sql, params) {
Ok(n) => results.push(n),
Err(e) => {
error = Some(e);
break;
}
}
}
async move {
match error {
Some(e) => Outcome::Err(e),
None => Outcome::Ok(results),
}
}
}
fn begin(&self, cx: &Cx) -> impl Future<Output = Outcome<Self::Tx<'_>, Error>> + Send {
self.begin_with(cx, IsolationLevel::default())
}
fn begin_with(
&self,
_cx: &Cx,
isolation: IsolationLevel,
) -> impl Future<Output = Outcome<Self::Tx<'_>, Error>> + Send {
let result = self
.begin_sync(isolation)
.map(|()| SqliteTransaction::new(self));
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn prepare(
&self,
_cx: &Cx,
sql: &str,
) -> impl Future<Output = Outcome<PreparedStatement, Error>> + Send {
let inner = self.inner.lock().unwrap_or_else(|e| e.into_inner());
let result = prepare_stmt(inner.db, sql).map(|stmt| {
let param_count = unsafe { ffi::sqlite3_bind_parameter_count(stmt) } as usize;
let col_count = unsafe { ffi::sqlite3_column_count(stmt) } as c_int;
let mut columns = Vec::with_capacity(col_count as usize);
for i in 0..col_count {
if let Some(name) = unsafe { types::column_name(stmt, i) } {
columns.push(name);
}
}
unsafe { ffi::sqlite3_finalize(stmt) };
let id = sql.as_ptr() as u64;
PreparedStatement::with_columns(id, sql.to_string(), param_count, columns)
});
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn query_prepared(
&self,
cx: &Cx,
stmt: &PreparedStatement,
params: &[Value],
) -> impl Future<Output = Outcome<Vec<Row>, Error>> + Send {
self.query(cx, stmt.sql(), params)
}
fn execute_prepared(
&self,
cx: &Cx,
stmt: &PreparedStatement,
params: &[Value],
) -> impl Future<Output = Outcome<u64, Error>> + Send {
self.execute(cx, stmt.sql(), params)
}
fn ping(&self, _cx: &Cx) -> impl Future<Output = Outcome<(), Error>> + Send {
let result = self.query_sync("SELECT 1", &[]).map(|_| ());
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn close(self, _cx: &Cx) -> impl Future<Output = sqlmodel_core::Result<()>> + Send {
std::future::ready(Ok(()))
}
}
impl TransactionOps for SqliteTransaction<'_> {
fn query(
&self,
_cx: &Cx,
sql: &str,
params: &[Value],
) -> impl Future<Output = Outcome<Vec<Row>, Error>> + Send {
let result = self.conn.query_sync(sql, params);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn query_one(
&self,
_cx: &Cx,
sql: &str,
params: &[Value],
) -> impl Future<Output = Outcome<Option<Row>, Error>> + Send {
let result = self.conn.query_sync(sql, params).map(|mut rows| rows.pop());
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn execute(
&self,
_cx: &Cx,
sql: &str,
params: &[Value],
) -> impl Future<Output = Outcome<u64, Error>> + Send {
let result = self.conn.execute_sync(sql, params);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn savepoint(&self, _cx: &Cx, name: &str) -> impl Future<Output = Outcome<(), Error>> + Send {
let quoted_name = format!("\"{}\"", name.replace('"', "\"\""));
let sql = format!("SAVEPOINT {}", quoted_name);
let result = self.conn.execute_raw(&sql);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn rollback_to(&self, _cx: &Cx, name: &str) -> impl Future<Output = Outcome<(), Error>> + Send {
let quoted_name = format!("\"{}\"", name.replace('"', "\"\""));
let sql = format!("ROLLBACK TO {}", quoted_name);
let result = self.conn.execute_raw(&sql);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn release(&self, _cx: &Cx, name: &str) -> impl Future<Output = Outcome<(), Error>> + Send {
let quoted_name = format!("\"{}\"", name.replace('"', "\"\""));
let sql = format!("RELEASE {}", quoted_name);
let result = self.conn.execute_raw(&sql);
async move { result.map_or_else(Outcome::Err, Outcome::Ok) }
}
fn commit(mut self, _cx: &Cx) -> impl Future<Output = Outcome<(), Error>> + Send {
self.committed = true;
std::future::ready(
self.conn
.commit_sync()
.map_or_else(Outcome::Err, Outcome::Ok),
)
}
fn rollback(mut self, _cx: &Cx) -> impl Future<Output = Outcome<(), Error>> + Send {
self.committed = true; std::future::ready(
self.conn
.rollback_sync()
.map_or_else(Outcome::Err, Outcome::Ok),
)
}
}
fn connection_uses_shared_cache(config: &SqliteConfig) -> bool {
if config.path == ":memory:" || config.path.is_empty() {
return false;
}
let uri_mode = sqlite_uri_cache_mode(&config.path);
if config.flags.uri {
uri_mode.unwrap_or(config.flags.shared_cache)
} else {
config.flags.shared_cache || uri_mode == Some(true)
}
}
struct BackupBusyTimeoutGuard {
source_db: *mut ffi::sqlite3,
dest_db: *mut ffi::sqlite3,
source_timeout_ms: c_int,
dest_timeout_ms: c_int,
}
impl BackupBusyTimeoutGuard {
fn disable(
source_db: *mut ffi::sqlite3,
dest_db: *mut ffi::sqlite3,
source_timeout_ms: c_int,
dest_timeout_ms: c_int,
) -> Self {
let source_rc = unsafe { ffi::sqlite3_busy_timeout(source_db, 0) };
let dest_rc = unsafe { ffi::sqlite3_busy_timeout(dest_db, 0) };
debug_assert_eq!(source_rc, ffi::SQLITE_OK);
debug_assert_eq!(dest_rc, ffi::SQLITE_OK);
Self {
source_db,
dest_db,
source_timeout_ms,
dest_timeout_ms,
}
}
}
fn sqlite_busy_timeout_ms(db: *mut ffi::sqlite3) -> Result<c_int, Error> {
const SQL: &str = "PRAGMA busy_timeout";
let stmt = prepare_stmt(db, SQL)?;
let row_rc = unsafe { ffi::sqlite3_step(stmt) };
if row_rc != ffi::SQLITE_ROW {
let error = step_error(db, SQL, row_rc);
unsafe { ffi::sqlite3_finalize(stmt) };
return Err(error);
}
let timeout_ms = unsafe { ffi::sqlite3_column_int(stmt, 0) };
let done_rc = unsafe { ffi::sqlite3_step(stmt) };
if done_rc != ffi::SQLITE_DONE {
let error = step_error(db, SQL, done_rc);
unsafe { ffi::sqlite3_finalize(stmt) };
return Err(error);
}
unsafe { ffi::sqlite3_finalize(stmt) };
Ok(timeout_ms.max(0))
}
impl Drop for BackupBusyTimeoutGuard {
fn drop(&mut self) {
let source_rc =
unsafe { ffi::sqlite3_busy_timeout(self.source_db, self.source_timeout_ms) };
let dest_rc = unsafe { ffi::sqlite3_busy_timeout(self.dest_db, self.dest_timeout_ms) };
debug_assert_eq!(source_rc, ffi::SQLITE_OK);
debug_assert_eq!(dest_rc, ffi::SQLITE_OK);
}
}
fn sqlite_uri_cache_mode(path: &str) -> Option<bool> {
let query = path.strip_prefix("file:")?.split_once('?')?.1;
let query = query.split_once('#').map_or(query, |(query, _)| query);
let mut cache_mode = None;
for parameter in query.split('&') {
let (name, value) = parameter.split_once('=').unwrap_or((parameter, ""));
let name = percent_decode_uri_component(name);
if name != b"cache" {
continue;
}
match percent_decode_uri_component(value).as_slice() {
b"shared" => cache_mode = Some(true),
b"private" => cache_mode = Some(false),
_ => {
}
}
}
cache_mode
}
fn percent_decode_uri_component(component: &str) -> Vec<u8> {
let bytes = component.as_bytes();
let mut decoded = Vec::with_capacity(bytes.len());
let mut index = 0;
while index < bytes.len() {
if bytes[index] == b'%' {
if let (Some(high), Some(low)) = (bytes.get(index + 1), bytes.get(index + 2)) {
if let (Some(high), Some(low)) = (hex_nibble(*high), hex_nibble(*low)) {
let byte = (high << 4) | low;
if byte == 0 {
break;
}
decoded.push(byte);
index += 3;
continue;
}
}
}
decoded.push(bytes[index]);
index += 1;
}
decoded
}
const fn hex_nibble(byte: u8) -> Option<u8> {
match byte {
b'0'..=b'9' => Some(byte - b'0'),
b'a'..=b'f' => Some(byte - b'a' + 10),
b'A'..=b'F' => Some(byte - b'A' + 10),
_ => None,
}
}
fn sqlite_error_code_from_db(db: *mut ffi::sqlite3, result: c_int) -> SqliteErrorCode {
let primary = result & 0xff;
let observed_extended = if db.is_null() {
result
} else {
unsafe { ffi::sqlite3_extended_errcode(db) }
};
let extended = if observed_extended & 0xff == primary {
observed_extended
} else {
result
};
SqliteErrorCode::from_result_codes(result, extended)
}
fn direct_sqlite_error_code(result: c_int) -> SqliteErrorCode {
SqliteErrorCode::from_result_codes(result, result)
}
fn backup_step_error(db: *mut ffi::sqlite3, result: c_int) -> Error {
let detail = if db.is_null() {
ffi::error_string(result).to_string()
} else {
unsafe { CStr::from_ptr(ffi::sqlite3_errmsg(db)) }
.to_string_lossy()
.into_owned()
};
Error::Connection(ConnectionError {
kind: ConnectionErrorKind::Connect,
message: format!(
"SQLite backup failed: {detail} ({})",
ffi::error_string(result)
),
source: Some(Box::new(direct_sqlite_error_code(result))),
})
}
fn prepare_stmt(db: *mut ffi::sqlite3, sql: &str) -> Result<*mut ffi::sqlite3_stmt, Error> {
let c_sql = CString::new(sql).map_err(|_| {
Error::Query(QueryError {
kind: QueryErrorKind::Syntax,
sql: Some(sql.to_string()),
sqlstate: None,
message: "SQL contains null byte".to_string(),
detail: None,
hint: None,
position: None,
source: None,
})
})?;
let mut stmt: *mut ffi::sqlite3_stmt = ptr::null_mut();
let rc = unsafe {
ffi::sqlite3_prepare_v2(
db,
c_sql.as_ptr(),
c_sql.as_bytes().len() as c_int,
&mut stmt,
ptr::null_mut(),
)
};
if rc != ffi::SQLITE_OK {
return Err(prepare_error(db, sql, rc));
}
if stmt.is_null() {
return Err(Error::Query(QueryError {
kind: QueryErrorKind::Syntax,
sql: Some(sql.to_string()),
sqlstate: None,
message: "SQL contains no executable statement".to_string(),
detail: None,
hint: None,
position: None,
source: None,
}));
}
Ok(stmt)
}
fn prepare_error(db: *mut ffi::sqlite3, sql: &str, code: c_int) -> Error {
let msg = unsafe {
let ptr = ffi::sqlite3_errmsg(db);
CStr::from_ptr(ptr).to_string_lossy().into_owned()
};
let error_code = sqlite_error_code_from_db(db, code);
Error::Query(QueryError {
kind: error_code_to_kind(code),
sql: Some(sql.to_string()),
sqlstate: None,
message: msg,
detail: None,
hint: None,
position: None,
source: Some(Box::new(error_code)),
})
}
fn bind_error(db: *mut ffi::sqlite3, sql: &str, param_index: usize, code: c_int) -> Error {
let msg = unsafe {
let ptr = ffi::sqlite3_errmsg(db);
CStr::from_ptr(ptr).to_string_lossy().into_owned()
};
let error_code = sqlite_error_code_from_db(db, code);
Error::Query(QueryError {
kind: error_code_to_kind(code),
sql: Some(sql.to_string()),
sqlstate: None,
message: format!("Failed to bind parameter {}: {}", param_index, msg),
detail: None,
hint: None,
position: None,
source: Some(Box::new(error_code)),
})
}
fn step_error(db: *mut ffi::sqlite3, sql: &str, code: c_int) -> Error {
let msg = unsafe {
let ptr = ffi::sqlite3_errmsg(db);
CStr::from_ptr(ptr).to_string_lossy().into_owned()
};
let error_code = sqlite_error_code_from_db(db, code);
Error::Query(QueryError {
kind: error_code_to_kind(code),
sql: Some(sql.to_string()),
sqlstate: None,
message: msg,
detail: None,
hint: None,
position: None,
source: Some(Box::new(error_code)),
})
}
fn error_code_to_kind(code: c_int) -> QueryErrorKind {
match code & 0xff {
ffi::SQLITE_CONSTRAINT => QueryErrorKind::Constraint,
ffi::SQLITE_BUSY | ffi::SQLITE_LOCKED => QueryErrorKind::Deadlock,
ffi::SQLITE_PERM | ffi::SQLITE_READONLY | ffi::SQLITE_AUTH => QueryErrorKind::Permission,
ffi::SQLITE_NOTFOUND => QueryErrorKind::NotFound,
ffi::SQLITE_TOOBIG => QueryErrorKind::DataTruncation,
ffi::SQLITE_INTERRUPT => QueryErrorKind::Cancelled,
_ => QueryErrorKind::Database,
}
}
#[allow(dead_code)]
fn format_value(value: &Value) -> String {
match value {
Value::Null => "NULL".to_string(),
Value::Bool(b) => if *b { "true" } else { "false" }.to_string(),
Value::TinyInt(n) => n.to_string(),
Value::SmallInt(n) => n.to_string(),
Value::Int(n) => n.to_string(),
Value::BigInt(n) => n.to_string(),
Value::Float(n) => format!("{:.6}", n),
Value::Double(n) => format!("{:.6}", n),
Value::Text(s) => s.clone(),
Value::Bytes(b) => format!("[BLOB: {} bytes]", b.len()),
Value::Date(d) => d.to_string(),
Value::Time(t) => t.to_string(),
Value::Timestamp(ts) => ts.to_string(),
Value::TimestampTz(ts) => ts.to_string(),
Value::Json(j) => j.to_string(),
Value::Uuid(u) => {
format!(
"{:02x}{:02x}{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}",
u[0],
u[1],
u[2],
u[3],
u[4],
u[5],
u[6],
u[7],
u[8],
u[9],
u[10],
u[11],
u[12],
u[13],
u[14],
u[15]
)
}
Value::Decimal(d) => d.to_string(),
Value::Array(arr) => format!("[{} items]", arr.len()),
Value::Default => "DEFAULT".to_string(),
}
}
#[cfg(feature = "console")]
impl ConsoleAware for SqliteConnection {
fn set_console(&mut self, console: Option<Arc<SqlModelConsole>>) {
self.console = console;
self.emit_open_status();
}
fn console(&self) -> Option<&Arc<SqlModelConsole>> {
self.console.as_ref()
}
fn has_console(&self) -> bool {
self.console.is_some()
}
}
impl SqliteConnection {
#[cfg(feature = "console")]
fn emit_open_status(&self) {
if let Some(console) = &self.console {
let mode = if self.path == ":memory:" {
"in-memory"
} else {
"file"
};
let journal_mode = self
.query_sync("PRAGMA journal_mode", &[])
.ok()
.and_then(|rows| rows.first().and_then(|r| r.get_as::<String>(0).ok()));
let page_size = self
.query_sync("PRAGMA page_size", &[])
.ok()
.and_then(|rows| rows.first().and_then(|r| r.get_as::<i64>(0).ok()));
if console.mode().is_plain() {
let journal = journal_mode.as_deref().unwrap_or("unknown");
console.status(&format!(
"Opened SQLite database: {} ({} mode, journal: {})",
self.path, mode, journal
));
} else {
console.status(&format!("SQLite database: {}", self.path));
console.status(&format!(" Mode: {}", mode));
if let Some(journal) = journal_mode {
console.status(&format!(" Journal: {}", journal.to_uppercase()));
}
if let Some(size) = page_size {
console.status(&format!(" Page size: {} bytes", size));
}
}
}
}
#[cfg(feature = "console")]
fn emit_transaction_state(&self, state: &str) {
if let Some(console) = &self.console {
if console.mode().is_plain() {
console.status(&format!("Transaction: {}", state));
} else {
console.status(&format!("[{}] Transaction {}", state, state.to_lowercase()));
}
}
}
#[cfg(feature = "console")]
fn emit_query_timing(&self, elapsed_ms: f64, rows: usize) {
if let Some(console) = &self.console {
console.status(&format!("Query: {:.1}ms, {} rows", elapsed_ms, rows));
}
}
#[cfg(feature = "console")]
fn emit_query_result(&self, sql: &str, col_names: &[String], rows: &[Row], elapsed_ms: f64) {
if let Some(console) = &self.console {
let sql_upper = sql.trim().to_uppercase();
let is_pragma = sql_upper.starts_with("PRAGMA");
if is_pragma && !rows.is_empty() {
if console.mode().is_plain() {
console.status(&format!("{}:", sql.trim()));
console.status(&format!(" {}", col_names.join("|")));
for row in rows.iter().take(20) {
let values: Vec<String> = (0..col_names.len())
.map(|i| {
row.get(i)
.map(|v| format_value(v))
.unwrap_or_else(|| "NULL".to_string())
})
.collect();
console.status(&format!(" {}", values.join("|")));
}
if rows.len() > 20 {
console.status(&format!(" ... and {} more rows", rows.len() - 20));
}
console.status(&format!(" ({:.1}ms)", elapsed_ms));
} else {
let mut table_output = String::new();
table_output.push_str(&format!("PRAGMA Query Results ({:.1}ms)\n", elapsed_ms));
let mut widths: Vec<usize> = col_names.iter().map(|c| c.len()).collect();
for row in rows.iter().take(20) {
for (i, w) in widths.iter_mut().enumerate() {
let val_len = row.get(i).map(|v| format_value(v).len()).unwrap_or(4); if val_len > *w {
*w = val_len;
}
}
}
let sep: String = widths
.iter()
.map(|w| "-".repeat(*w + 2))
.collect::<Vec<_>>()
.join("+");
table_output.push_str(&format!("+{}+\n", sep));
let header: String = col_names
.iter()
.enumerate()
.map(|(i, name)| format!(" {:width$} ", name, width = widths[i]))
.collect::<Vec<_>>()
.join("|");
table_output.push_str(&format!("|{}|\n", header));
table_output.push_str(&format!("+{}+\n", sep));
for row in rows.iter().take(20) {
let data: String = (0..col_names.len())
.map(|i| {
let val = row
.get(i)
.map(|v| format_value(v))
.unwrap_or_else(|| "NULL".to_string());
format!(" {:width$} ", val, width = widths[i])
})
.collect::<Vec<_>>()
.join("|");
table_output.push_str(&format!("|{}|\n", data));
}
table_output.push_str(&format!("+{}+", sep));
if rows.len() > 20 {
table_output.push_str(&format!("\n... and {} more rows", rows.len() - 20));
}
console.status(&table_output);
}
} else {
self.emit_query_timing(elapsed_ms, rows.len());
}
}
}
#[cfg(feature = "console")]
fn emit_execute_timing(&self, sql: &str, rows_affected: u64, elapsed_ms: f64) {
if let Some(console) = &self.console {
let sql_upper = sql.trim().to_uppercase();
let op_type = if sql_upper.starts_with("INSERT") {
"Insert"
} else if sql_upper.starts_with("UPDATE") {
"Update"
} else if sql_upper.starts_with("DELETE") {
"Delete"
} else if sql_upper.starts_with("CREATE") {
"Create"
} else if sql_upper.starts_with("DROP") {
"Drop"
} else if sql_upper.starts_with("ALTER") {
"Alter"
} else {
"Execute"
};
if console.mode().is_plain() {
console.status(&format!(
"{}: {} rows affected ({:.1}ms)",
op_type, rows_affected, elapsed_ms
));
} else {
console.status(&format!(
"[{}] {} rows affected ({:.1}ms)",
op_type.to_uppercase(),
rows_affected,
elapsed_ms
));
}
}
}
#[cfg(feature = "console")]
pub fn emit_busy_waiting(&self, elapsed_secs: f64) {
if let Some(console) = &self.console {
if console.mode().is_plain() {
console.status(&format!(
"Waiting for database lock... ({:.1}s)",
elapsed_secs
));
} else {
console.status(&format!(
"[..] Waiting for database lock... ({:.1}s)",
elapsed_secs
));
}
}
}
#[cfg(feature = "console")]
pub fn emit_checkpoint_progress(&self, pages_done: u32, pages_total: u32) {
if let Some(console) = &self.console {
let pct = if pages_total > 0 {
(pages_done as f64 / pages_total as f64) * 100.0
} else {
100.0
};
if console.mode().is_plain() {
console.status(&format!(
"WAL checkpoint: {:.0}% ({}/{} pages)",
pct, pages_done, pages_total
));
} else {
let bar_width: usize = 20;
let filled = ((pct / 100.0) * bar_width as f64).round() as usize;
let empty = bar_width.saturating_sub(filled);
let bar = format!("[{}{}]", "=".repeat(filled), " ".repeat(empty));
console.status(&format!(
"WAL checkpoint: {} {:.0}% ({}/{} pages)",
bar, pct, pages_done, pages_total
));
}
}
}
#[cfg(not(feature = "console"))]
#[allow(dead_code)]
fn emit_open_status(&self) {}
#[cfg(not(feature = "console"))]
fn emit_transaction_state(&self, _state: &str) {}
#[cfg(not(feature = "console"))]
#[allow(dead_code)]
fn emit_query_timing(&self, _elapsed_ms: f64, _rows: usize) {}
#[cfg(not(feature = "console"))]
#[allow(dead_code)]
fn emit_query_result(
&self,
_sql: &str,
_col_names: &[String],
_rows: &[Row],
_elapsed_ms: f64,
) {
}
#[cfg(not(feature = "console"))]
#[allow(dead_code)]
fn emit_execute_timing(&self, _sql: &str, _rows_affected: u64, _elapsed_ms: f64) {}
#[cfg(not(feature = "console"))]
pub fn emit_busy_waiting(&self, _elapsed_secs: f64) {}
#[cfg(not(feature = "console"))]
pub fn emit_checkpoint_progress(&self, _pages_done: u32, _pages_total: u32) {}
}
#[cfg(test)]
mod tests {
use super::*;
static NEXT_TEMP_DB: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
fn unique_temp_db_path(label: &str) -> std::path::PathBuf {
let nonce = NEXT_TEMP_DB.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
std::env::temp_dir().join(format!(
"sqlmodel_{label}_{}_{}.db",
std::process::id(),
nonce
))
}
#[test]
fn test_open_memory() {
let conn = SqliteConnection::open_memory().unwrap();
assert_eq!(conn.path(), ":memory:");
}
#[test]
fn test_execute_raw() {
let conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)")
.unwrap();
conn.execute_raw("INSERT INTO test (name) VALUES ('Alice')")
.unwrap();
assert_eq!(conn.changes(), 1);
assert_eq!(conn.last_insert_rowid(), 1);
let pre_sqlite_error = conn
.execute_raw("SELECT \0")
.expect_err("NUL-bearing SQL must fail before SQLite");
assert_eq!(
sqlite_error_code(&pre_sqlite_error),
None,
"errors produced before the native call must not invent a SQLite result code"
);
}
#[test]
fn test_query_sync() {
let conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)")
.unwrap();
conn.execute_raw("INSERT INTO test (name) VALUES ('Alice'), ('Bob')")
.unwrap();
let rows = conn
.query_sync("SELECT * FROM test ORDER BY id", &[])
.unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(rows[0].get_named::<i32>("id").unwrap(), 1);
assert_eq!(rows[0].get_named::<String>("name").unwrap(), "Alice");
assert_eq!(rows[1].get_named::<i32>("id").unwrap(), 2);
assert_eq!(rows[1].get_named::<String>("name").unwrap(), "Bob");
}
#[test]
fn test_parameterized_query() {
let conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
.unwrap();
conn.execute_sync(
"INSERT INTO test (name, age) VALUES (?, ?)",
&[Value::Text("Alice".to_string()), Value::Int(30)],
)
.unwrap();
let rows = conn
.query_sync(
"SELECT * FROM test WHERE name = ?",
&[Value::Text("Alice".to_string())],
)
.unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get_named::<String>("name").unwrap(), "Alice");
assert_eq!(rows[0].get_named::<i32>("age").unwrap(), 30);
}
#[test]
fn test_prepared_errors_retain_exact_native_codes() {
let conn = SqliteConnection::open_memory().unwrap();
let prepare_error = conn
.query_sync("SELEC 1", &[])
.expect_err("invalid SQL must fail during prepare");
let prepare_code = sqlite_error_code(&prepare_error)
.expect("prepare failures must retain their native SQLite result code");
assert_eq!(prepare_code.primary(), ffi::SQLITE_ERROR);
assert_eq!(prepare_code.extended(), ffi::SQLITE_ERROR);
let query_bind_error = conn
.query_sync("SELECT ?1", &[Value::Int(1), Value::Int(2)])
.expect_err("binding beyond the statement parameter count must fail");
let execute_bind_error = conn
.execute_sync("SELECT ?1", &[Value::Int(1), Value::Int(2)])
.expect_err("execute_sync must retain the same bind failure");
for error in [&query_bind_error, &execute_bind_error] {
let code = sqlite_error_code(error)
.expect("bind failures must survive statement finalization");
assert_eq!(code.primary(), ffi::SQLITE_RANGE);
assert_eq!(code.extended(), ffi::SQLITE_RANGE);
}
conn.execute_raw("CREATE TABLE exact_codes (value INTEGER UNIQUE)")
.unwrap();
conn.execute_sync("INSERT INTO exact_codes VALUES (1)", &[])
.unwrap();
let execute_step_error = conn
.execute_sync("INSERT INTO exact_codes VALUES (1)", &[])
.expect_err("duplicate prepared insert must fail during step");
let query_step_error = conn
.query_sync("INSERT INTO exact_codes VALUES (1) RETURNING value", &[])
.expect_err("query_sync must retain a step failure before finalization");
for error in [&execute_step_error, &query_step_error] {
let code = sqlite_error_code(error)
.expect("step failures must retain their extended SQLite result code");
assert_eq!(code.primary(), ffi::SQLITE_CONSTRAINT);
assert_eq!(code.extended(), ffi::SQLITE_CONSTRAINT_UNIQUE);
assert!(
matches!(error, Error::Query(query) if query.kind == QueryErrorKind::Constraint),
"unique violations should map to the constraint error family: {error}"
);
}
}
#[test]
fn test_empty_prepared_sql_is_rejected_before_statement_ffi() {
let conn = SqliteConnection::open_memory().unwrap();
for sql in ["", " \n\t", "-- comment only\n", "/* comment only */"] {
for error in [
conn.query_sync(sql, &[])
.expect_err("empty query SQL must not produce a null statement"),
conn.execute_sync(sql, &[])
.expect_err("empty execute SQL must not produce a null statement"),
] {
assert!(
matches!(error, Error::Query(ref query) if query.kind == QueryErrorKind::Syntax),
"empty prepared SQL should be a typed syntax error: {error}"
);
assert!(error.to_string().contains("no executable statement"));
assert_eq!(sqlite_error_code(&error), None);
}
}
}
#[test]
fn test_execute_returning_steps_until_done_and_retains_late_busy() {
let path = unique_temp_db_path("returning_busy");
let _ = std::fs::remove_file(&path);
let config = SqliteConfig::file(path.to_string_lossy().into_owned()).busy_timeout(0);
let writer = SqliteConnection::open(&config).unwrap();
writer.execute_raw("PRAGMA journal_mode=DELETE").unwrap();
writer
.execute_raw("CREATE TABLE returning_rows (value INTEGER)")
.unwrap();
let reader = SqliteConnection::open(&config).unwrap();
reader.execute_raw("BEGIN DEFERRED").unwrap();
reader
.query_sync("SELECT COUNT(*) FROM returning_rows", &[])
.unwrap();
let error = writer
.execute_sync("INSERT INTO returning_rows VALUES (7) RETURNING value", &[])
.expect_err("the read lock must surface after RETURNING rows but before DONE");
let code = sqlite_error_code(&error)
.expect("late RETURNING completion failure must retain its native code");
assert_eq!(code.primary(), ffi::SQLITE_BUSY);
assert!(
matches!(error, Error::Query(ref query) if query.kind == QueryErrorKind::Deadlock),
"late SQLITE_BUSY should map to the deadlock family: {error}"
);
reader.execute_raw("ROLLBACK").unwrap();
let rows = writer
.query_sync("SELECT COUNT(*) AS row_count FROM returning_rows", &[])
.unwrap();
assert_eq!(rows[0].get_named::<i64>("row_count").unwrap(), 0);
drop(reader);
drop(writer);
let _ = std::fs::remove_file(path);
}
#[test]
fn test_non_native_connection_preflights_have_no_sqlite_code() {
let invalid_timeout = SqliteConfig::memory().busy_timeout(c_int::MAX as u32 + 1);
let Err(timeout_error) = SqliteConnection::open(&invalid_timeout) else {
panic!("out-of-range native busy timeout must fail closed");
};
assert!(matches!(timeout_error, Error::Config(_)));
assert_eq!(sqlite_error_code(&timeout_error), None);
let conn = SqliteConnection::open_memory().unwrap();
let backup_error = conn
.backup_to_connection(&conn)
.expect_err("backing a connection up onto itself must fail before locking");
assert!(
backup_error
.to_string()
.contains("source and destination must be different")
);
assert_eq!(sqlite_error_code(&backup_error), None);
let contradictory_cache_flags = SqliteConfig::memory().flags(OpenFlags {
shared_cache: true,
private_cache: true,
..OpenFlags::create_read_write()
});
let Err(cache_error) = SqliteConnection::open(&contradictory_cache_flags) else {
panic!("contradictory SQLite cache flags must fail before native open");
};
assert!(matches!(cache_error, Error::Config(_)));
assert_eq!(sqlite_error_code(&cache_error), None);
assert_ne!(
OpenFlags::create_read_write().to_sqlite_flags() & ffi::SQLITE_OPEN_PRIVATECACHE,
0,
"ordinary opens must override process-global shared-cache mode"
);
}
#[test]
fn test_backup_copies_data_and_opposite_directions_do_not_deadlock() {
let left = Arc::new(SqliteConnection::open_memory().unwrap());
let right = Arc::new(SqliteConnection::open_memory().unwrap());
left.execute_raw("CREATE TABLE backup_rows (value INTEGER)")
.unwrap();
left.execute_raw("INSERT INTO backup_rows VALUES (7)")
.unwrap();
left.backup_to_connection(&right)
.expect("ordinary backup should copy the source database");
let copied = right
.query_sync("SELECT value FROM backup_rows", &[])
.expect("copied table should be readable");
assert_eq!(copied[0].get_named::<i64>("value").unwrap(), 7);
let barrier = Arc::new(std::sync::Barrier::new(3));
let (completed_tx, completed_rx) = std::sync::mpsc::channel();
let mut workers = Vec::new();
for (source, destination) in [
(Arc::clone(&left), Arc::clone(&right)),
(Arc::clone(&right), Arc::clone(&left)),
] {
let worker_barrier = Arc::clone(&barrier);
let worker_tx = completed_tx.clone();
workers.push(std::thread::spawn(move || {
worker_barrier.wait();
let result = source.backup_to_connection(&destination);
worker_tx.send(result).expect("test receiver remains live");
}));
}
drop(completed_tx);
barrier.wait();
for _ in 0..2 {
completed_rx
.recv_timeout(Duration::from_secs(5))
.expect("opposing backups must not deadlock")
.expect("serialized opposing backup should succeed");
}
for worker in workers {
worker.join().expect("backup worker should not panic");
}
}
#[test]
fn test_backup_lock_retries_respect_deadline_and_restore_busy_timeouts() {
let source = SqliteConnection::open(&SqliteConfig::memory().busy_timeout(500)).unwrap();
source
.execute_raw("CREATE TABLE backup_rows (value INTEGER)")
.unwrap();
source
.execute_raw("INSERT INTO backup_rows VALUES (7)")
.unwrap();
let path = unique_temp_db_path("backup_deadline");
let _ = std::fs::remove_file(&path);
let destination_config =
SqliteConfig::file(path.to_string_lossy().into_owned()).busy_timeout(450);
let destination = SqliteConnection::open(&destination_config).unwrap();
destination
.execute_raw("CREATE TABLE old_rows (value INTEGER)")
.unwrap();
source.execute_raw("PRAGMA busy_timeout=520").unwrap();
destination.execute_raw("PRAGMA busy_timeout=470").unwrap();
let blocker = SqliteConnection::open(&destination_config).unwrap();
blocker.execute_raw("BEGIN EXCLUSIVE").unwrap();
let started = Instant::now();
let error = source
.backup_to_connection(&destination)
.expect_err("an exclusive destination lock must block the backup");
let elapsed = started.elapsed();
let code = sqlite_error_code(&error)
.expect("a lock-blocked backup must retain its native SQLite result code");
assert!(
matches!(code.primary(), ffi::SQLITE_BUSY | ffi::SQLITE_LOCKED),
"unexpected lock failure code: {code}"
);
assert!(
elapsed < Duration::from_millis(800),
"backup retry deadline overran by a native busy-timeout interval: {elapsed:?}"
);
blocker.execute_raw("ROLLBACK").unwrap();
for (connection, expected_timeout) in [(&source, 520), (&destination, 470)] {
let rows = connection.query_sync("PRAGMA busy_timeout", &[]).unwrap();
assert_eq!(
rows[0].get_named::<i32>("timeout").unwrap(),
expected_timeout
);
}
drop(blocker);
drop(destination);
let _ = std::fs::remove_file(path);
}
#[test]
fn test_backup_rejects_shared_cache_destination_before_locking() {
let source = SqliteConnection::open_memory().unwrap();
let shared_uri = format!(
"file:sqlmodel_backup_shared_{}?mode=memory&cache=shared",
std::process::id()
);
let destination =
SqliteConnection::open(&SqliteConfig::file(shared_uri).flags(OpenFlags {
uri: true,
..OpenFlags::create_read_write()
}))
.expect("shared-cache connection should open for the preflight test");
assert!(destination.uses_shared_cache);
let error = source
.backup_to_connection(&destination)
.expect_err("shared-cache backup destination must fail closed");
assert!(error.to_string().contains("shared-cache mode"));
assert_eq!(sqlite_error_code(&error), None);
let plain_memory = SqliteConnection::open(&SqliteConfig::memory().flags(OpenFlags {
shared_cache: true,
..OpenFlags::create_read_write()
}))
.expect("plain :memory: remains private even with SHAREDCACHE requested");
assert!(!plain_memory.uses_shared_cache);
source
.backup_to_connection(&plain_memory)
.expect("a truly private in-memory destination is backup-safe");
let private_uri = format!(
"file:sqlmodel_backup_private_{}?mode=memory&cache=private",
std::process::id()
);
let uri_overrides_flag =
SqliteConnection::open(&SqliteConfig::file(private_uri).flags(OpenFlags {
uri: true,
shared_cache: true,
..OpenFlags::create_read_write()
}))
.expect("URI cache=private should override the shared-cache open flag");
assert!(!uri_overrides_flag.uses_shared_cache);
source
.backup_to_connection(&uri_overrides_flag)
.expect("an effectively private URI destination is backup-safe");
}
#[test]
fn test_sqlite_uri_cache_mode_matches_sqlite_uri_rules() {
assert_eq!(
sqlite_uri_cache_mode("file:memory?mode=memory&cache=shared"),
Some(true)
);
assert_eq!(
sqlite_uri_cache_mode("file:memory?cache=private&cache=shared"),
Some(true),
"SQLite applies duplicate cache parameters in order, so the last one wins"
);
assert_eq!(
sqlite_uri_cache_mode("file:memory?%63ache=%70rivate"),
Some(false),
"SQLite percent-decodes URI parameter names and values"
);
assert_eq!(
sqlite_uri_cache_mode("file:memory?cache%00ignored=shared%00ignored"),
Some(true),
"SQLite truncates URI components at encoded NUL bytes"
);
assert_eq!(
sqlite_uri_cache_mode("file:memory?CACHE=shared"),
None,
"SQLite URI parameter names are case-sensitive"
);
assert_eq!(
sqlite_uri_cache_mode("file:memory?cache=shared#cache=private"),
Some(true),
"SQLite ignores URI fragments"
);
}
#[test]
fn test_backup_failure_retains_native_destination_error() {
let source = SqliteConnection::open_memory().unwrap();
source
.execute_raw("CREATE TABLE backup_source (value INTEGER)")
.unwrap();
let path = unique_temp_db_path("readonly_backup");
let writable = SqliteConnection::open_file(path.to_string_lossy().into_owned()).unwrap();
writable
.execute_raw("CREATE TABLE backup_destination (value INTEGER)")
.unwrap();
drop(writable);
let destination = SqliteConnection::open(
&SqliteConfig::file(path.to_string_lossy().into_owned()).flags(OpenFlags::read_only()),
)
.unwrap();
let error = source
.backup_to_connection(&destination)
.expect_err("read-only destination must reject backup writes");
let code = sqlite_error_code(&error)
.expect("native backup failure must retain its SQLite result code");
assert_eq!(code.primary(), ffi::SQLITE_READONLY);
assert!(error.to_string().to_ascii_lowercase().contains("readonly"));
drop(destination);
let _ = std::fs::remove_file(path);
}
#[test]
fn test_backup_step_error_preserves_direct_extended_result() {
let direct_extended = ffi::SQLITE_IOERR | (42 << 8);
let error = backup_step_error(std::ptr::null_mut(), direct_extended);
let code = sqlite_error_code(&error)
.expect("a backup-step failure must retain its direct native result");
assert_eq!(code.primary(), ffi::SQLITE_IOERR);
assert_eq!(code.extended(), direct_extended);
}
#[test]
fn test_null_handling() {
let conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)")
.unwrap();
conn.execute_sync("INSERT INTO test (name) VALUES (?)", &[Value::Null])
.unwrap();
let rows = conn.query_sync("SELECT * FROM test", &[]).unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get_named::<Option<String>>("name").unwrap(), None);
}
#[test]
fn test_transaction() {
let conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)")
.unwrap();
conn.begin_sync(IsolationLevel::default()).unwrap();
conn.execute_sync(
"INSERT INTO test (name) VALUES (?)",
&[Value::Text("Alice".to_string())],
)
.unwrap();
conn.rollback_sync().unwrap();
let rows = conn.query_sync("SELECT * FROM test", &[]).unwrap();
assert_eq!(rows.len(), 0);
conn.begin_sync(IsolationLevel::default()).unwrap();
conn.execute_sync(
"INSERT INTO test (name) VALUES (?)",
&[Value::Text("Bob".to_string())],
)
.unwrap();
conn.commit_sync().unwrap();
let rows = conn.query_sync("SELECT * FROM test", &[]).unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get_named::<String>("name").unwrap(), "Bob");
}
#[test]
fn test_insert_rowid() {
let conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)")
.unwrap();
let rowid = conn
.insert_sync(
"INSERT INTO test (name) VALUES (?)",
&[Value::Text("Alice".to_string())],
)
.unwrap();
assert_eq!(rowid, 1);
let rowid = conn
.insert_sync(
"INSERT INTO test (name) VALUES (?)",
&[Value::Text("Bob".to_string())],
)
.unwrap();
assert_eq!(rowid, 2);
}
#[test]
#[allow(clippy::approx_constant)]
fn test_type_conversions() {
let conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw(
"CREATE TABLE types (
b BOOLEAN,
i INTEGER,
f REAL,
t TEXT,
bl BLOB
)",
)
.unwrap();
conn.execute_sync(
"INSERT INTO types VALUES (?, ?, ?, ?, ?)",
&[
Value::Bool(true),
Value::BigInt(42),
Value::Double(3.14),
Value::Text("hello".to_string()),
Value::Bytes(vec![1, 2, 3]),
],
)
.unwrap();
let rows = conn.query_sync("SELECT * FROM types", &[]).unwrap();
assert_eq!(rows.len(), 1);
let b: i32 = rows[0].get_named("b").unwrap();
assert_eq!(b, 1);
let i: i32 = rows[0].get_named("i").unwrap();
assert_eq!(i, 42);
let f: f64 = rows[0].get_named("f").unwrap();
assert!((f - 3.14).abs() < 0.001);
let t: String = rows[0].get_named("t").unwrap();
assert_eq!(t, "hello");
let bl: Vec<u8> = rows[0].get_named("bl").unwrap();
assert_eq!(bl, vec![1, 2, 3]);
}
#[test]
fn test_open_flags() {
let tmp = unique_temp_db_path("open_flags");
let _ = std::fs::remove_file(&tmp);
let config = SqliteConfig::file(tmp.to_string_lossy().to_string())
.flags(OpenFlags::create_read_write());
let conn = SqliteConnection::open(&config).unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER)").unwrap();
drop(conn);
let config =
SqliteConfig::file(tmp.to_string_lossy().to_string()).flags(OpenFlags::read_only());
let conn = SqliteConnection::open(&config).unwrap();
let rows = conn.query_sync("SELECT * FROM test", &[]).unwrap();
assert_eq!(rows.len(), 0);
let error = conn
.execute_raw("INSERT INTO test VALUES (1)")
.expect_err("read-only connection must reject writes");
let error_code = sqlite_error_code(&error)
.expect("native write rejection must retain its exact SQLite result code");
assert_eq!(error_code.primary(), ffi::SQLITE_READONLY);
assert_eq!(error_code.extended() & 0xff, ffi::SQLITE_READONLY);
assert!(
matches!(error, Error::Query(ref query) if query.kind == QueryErrorKind::Permission),
"SQLITE_READONLY should map to the permission error family: {error}"
);
let prepared_error = conn
.execute_sync("INSERT INTO test VALUES (1)", &[])
.expect_err("prepared writes must also retain SQLITE_READONLY");
let prepared_code = sqlite_error_code(&prepared_error)
.expect("prepared write rejection must retain its native result code");
assert_eq!(prepared_code.primary(), ffi::SQLITE_READONLY);
assert!(
matches!(prepared_error, Error::Query(ref query) if query.kind == QueryErrorKind::Permission),
"prepared SQLITE_READONLY should map to permission: {prepared_error}"
);
drop(conn);
let _ = std::fs::remove_file(&tmp);
}
#[cfg(feature = "console")]
mod console_tests {
use super::*;
#[test]
fn test_console_aware_trait_impl() {
let mut conn = SqliteConnection::open_memory().unwrap();
assert!(!conn.has_console());
assert!(conn.console().is_none());
let console = Arc::new(SqlModelConsole::with_mode(
sqlmodel_console::OutputMode::Plain,
));
conn.set_console(Some(console.clone()));
assert!(conn.has_console());
assert!(conn.console().is_some());
conn.set_console(None);
assert!(!conn.has_console());
}
#[test]
fn test_database_open_feedback() {
let mut conn = SqliteConnection::open_memory().unwrap();
let console = Arc::new(SqlModelConsole::with_mode(
sqlmodel_console::OutputMode::Plain,
));
conn.set_console(Some(console));
}
#[test]
fn test_pragma_formatting() {
let mut conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)")
.unwrap();
let console = Arc::new(SqlModelConsole::with_mode(
sqlmodel_console::OutputMode::Plain,
));
conn.set_console(Some(console));
let rows = conn.query_sync("PRAGMA table_info(test)", &[]).unwrap();
assert!(!rows.is_empty());
}
#[test]
fn test_transaction_state() {
let mut conn = SqliteConnection::open_memory().unwrap();
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY)")
.unwrap();
let console = Arc::new(SqlModelConsole::with_mode(
sqlmodel_console::OutputMode::Plain,
));
conn.set_console(Some(console));
conn.begin_sync(IsolationLevel::default()).unwrap();
conn.execute_sync("INSERT INTO test (id) VALUES (?)", &[Value::Int(1)])
.unwrap();
conn.commit_sync().unwrap();
let rows = conn.query_sync("SELECT * FROM test", &[]).unwrap();
assert_eq!(rows.len(), 1);
}
#[test]
fn test_wal_checkpoint_progress() {
let conn = SqliteConnection::open_memory().unwrap();
conn.emit_checkpoint_progress(50, 100);
conn.emit_checkpoint_progress(100, 100);
conn.emit_checkpoint_progress(0, 0);
}
#[test]
fn test_busy_timeout_feedback() {
let conn = SqliteConnection::open_memory().unwrap();
conn.emit_busy_waiting(0.5);
conn.emit_busy_waiting(2.1);
}
#[test]
fn test_console_disabled_no_output() {
let conn = SqliteConnection::open_memory().unwrap();
conn.emit_busy_waiting(1.0);
conn.emit_checkpoint_progress(10, 100);
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY)")
.unwrap();
let rows = conn.query_sync("SELECT * FROM test", &[]).unwrap();
assert_eq!(rows.len(), 0);
}
#[test]
fn test_plain_mode_output() {
let mut conn = SqliteConnection::open_memory().unwrap();
let console = Arc::new(SqlModelConsole::with_mode(
sqlmodel_console::OutputMode::Plain,
));
conn.set_console(Some(console.clone()));
assert!(conn.console().unwrap().is_plain());
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)")
.unwrap();
conn.execute_sync(
"INSERT INTO test (name) VALUES (?)",
&[Value::Text("Alice".to_string())],
)
.unwrap();
let rows = conn.query_sync("PRAGMA table_info(test)", &[]).unwrap();
assert!(!rows.is_empty());
}
#[test]
fn test_rich_mode_output() {
let mut conn = SqliteConnection::open_memory().unwrap();
let console = Arc::new(SqlModelConsole::with_mode(
sqlmodel_console::OutputMode::Rich,
));
conn.set_console(Some(console.clone()));
assert!(conn.console().unwrap().is_rich());
conn.execute_raw("CREATE TABLE test (id INTEGER PRIMARY KEY)")
.unwrap();
conn.emit_checkpoint_progress(50, 100);
}
}
}