Lines
97.51 %
Functions
96.67 %
Branches
100 %
use std::{error::Error as StdError, sync::Arc};
use async_trait::async_trait;
use chrono::{TimeZone, Utc};
use futures::TryStreamExt;
use sql_builder::{quote, SqlBuilder};
use sqlx::SqlitePool;
use super::{
super::device::{
Cursor, Device, DeviceModel, ListOptions, ListQueryCond, QueryCond, SortKey,
UpdateQueryCond, Updates,
},
build_where_like,
};
/// Model instance.
pub struct Model {
/// The associated database connection.
conn: Arc<SqlitePool>,
}
/// Cursor instance.
///
/// The SQLite implementation uses the original list options and the progress offset.
pub struct DbCursor {
offset: u64,
/// SQLite schema.
#[derive(sqlx::FromRow)]
struct Schema {
device_id: String,
unit_id: String,
/// use empty string as NULL because duplicate
/// `(unit_code=NULL,network_code="code",network_addr="addr")` is allowed.
unit_code: String,
network_id: String,
network_code: String,
network_addr: String,
/// i64 as time tick from Epoch in milliseconds.
created_at: i64,
modified_at: i64,
profile: String,
name: String,
info: String,
/// Use "COUNT(*)" instead of "COUNT(fields...)" to simplify the implementation.
struct CountSchema {
#[sqlx(rename = "COUNT(*)")]
count: i64,
const TABLE_NAME: &'static str = "device";
const FIELDS: &'static [&'static str] = &[
"device_id",
"unit_id",
"unit_code",
"network_id",
"network_code",
"network_addr",
"created_at",
"modified_at",
"profile",
"name",
"info",
];
const TABLE_INIT_SQL: &'static str = "\
CREATE TABLE IF NOT EXISTS device (\
device_id TEXT NOT NULL UNIQUE,\
unit_id TEXT NOT NULL,\
unit_code TEXT NOT NULL,\
network_id TEXT NOT NULL,\
network_code TEXT NOT NULL,\
network_addr TEXT NOT NULL,\
created_at INTEGER NOT NULL,\
modified_at INTEGER NOT NULL,\
profile TEXT NOT NULL,\
name TEXT NOT NULL,\
info TEXT,\
UNIQUE (unit_code,network_code,network_addr),\
PRIMARY KEY (device_id))";
impl Model {
/// To create the model instance with a database connection.
pub async fn new(conn: Arc<SqlitePool>) -> Result<Self, Box<dyn StdError>> {
let model = Model { conn };
model.init().await?;
Ok(model)
#[async_trait]
impl DeviceModel for Model {
async fn init(&self) -> Result<(), Box<dyn StdError>> {
let _ = sqlx::query(TABLE_INIT_SQL)
.execute(self.conn.as_ref())
.await?;
Ok(())
async fn count(&self, cond: &ListQueryCond) -> Result<u64, Box<dyn StdError>> {
let sql = build_list_where(SqlBuilder::select_from(TABLE_NAME).count("*"), &cond).sql()?;
let result: Result<CountSchema, sqlx::Error> = sqlx::query_as(sql.as_str())
.fetch_one(self.conn.as_ref())
.await;
let row = match result {
Err(e) => return Err(Box::new(e)),
Ok(row) => row,
Ok(row.count as u64)
async fn list(
&self,
opts: &ListOptions,
cursor: Option<Box<dyn Cursor>>,
) -> Result<(Vec<Device>, Option<Box<dyn Cursor>>), Box<dyn StdError>> {
let mut cursor = match cursor {
None => Box::new(DbCursor::new()),
Some(cursor) => cursor,
let mut opts = ListOptions { ..*opts };
if let Some(offset) = opts.offset {
opts.offset = Some(offset + cursor.offset());
} else {
opts.offset = Some(cursor.offset());
let opts_limit = opts.limit;
if let Some(limit) = opts_limit {
if limit > 0 {
if cursor.offset() >= limit {
return Ok((vec![], None));
opts.limit = Some(limit - cursor.offset());
let mut builder = SqlBuilder::select_from(TABLE_NAME);
build_limit_offset(&mut builder, &opts);
build_sort(&mut builder, &opts);
let sql = build_list_where(&mut builder, opts.cond).sql()?;
let mut rows = sqlx::query_as::<_, Schema>(sql.as_str()).fetch(self.conn.as_ref());
let mut count: u64 = 0;
let mut list = vec![];
while let Some(row) = rows.try_next().await? {
let _ = cursor.as_mut().try_next().await?;
list.push(Device {
device_id: row.device_id,
unit_id: row.unit_id,
unit_code: match row.unit_code.len() {
0 => None,
_ => Some(row.unit_code),
network_id: row.network_id,
network_code: row.network_code,
network_addr: row.network_addr,
created_at: Utc.timestamp_nanos(row.created_at * 1000000),
modified_at: Utc.timestamp_nanos(row.modified_at * 1000000),
profile: row.profile,
name: row.name,
info: serde_json::from_str(row.info.as_str())?,
});
if limit > 0 && cursor.offset() >= limit {
if let Some(cursor_max) = opts.cursor_max {
if (count + 1) >= cursor_max {
return Ok((list, Some(cursor)));
return Ok((list, None));
count += 1;
if count >= cursor_max {
Ok((list, None))
async fn get(&self, cond: &QueryCond) -> Result<Option<Device>, Box<dyn StdError>> {
let sql = build_where(SqlBuilder::select_from(TABLE_NAME).fields(FIELDS), &cond).sql()?;
let result: Result<Schema, sqlx::Error> = sqlx::query_as(sql.as_str())
Err(e) => match e {
sqlx::Error::RowNotFound => return Ok(None),
_ => return Err(Box::new(e)),
Ok(Some(Device {
}))
async fn add(&self, device: &Device) -> Result<(), Box<dyn StdError>> {
let unit_code = match device.unit_code.as_deref() {
None => quote(""),
Some(value) => quote(value),
let info = match serde_json::to_string(&device.info) {
Err(_) => quote("{}"),
Ok(value) => quote(value.as_str()),
let values = vec![
quote(device.device_id.as_str()),
quote(device.unit_id.as_str()),
unit_code,
quote(device.network_id.as_str()),
quote(device.network_code.as_str()),
quote(device.network_addr.as_str()),
device.created_at.timestamp_millis().to_string(),
device.modified_at.timestamp_millis().to_string(),
quote(device.profile.as_str()),
quote(device.name.as_str()),
info,
let sql = SqlBuilder::insert_into(TABLE_NAME)
.fields(FIELDS)
.values(&values)
.sql()?;
let _ = sqlx::query(sql.as_str())
async fn add_bulk(&self, devices: &Vec<Device>) -> Result<(), Box<dyn StdError>> {
let mut builder = SqlBuilder::insert_into(TABLE_NAME);
builder.fields(FIELDS);
for device in devices.iter() {
builder.values(&vec![
]);
let sql = builder.sql()?.replace(");", ") ON CONFLICT DO NOTHING;");
async fn del(&self, cond: &QueryCond) -> Result<(), Box<dyn StdError>> {
let sql = build_where(&mut SqlBuilder::delete_from(TABLE_NAME), cond).sql()?;
async fn update(
cond: &UpdateQueryCond,
updates: &Updates,
) -> Result<(), Box<dyn StdError>> {
let sql = match build_update_where(&mut SqlBuilder::update_table(TABLE_NAME), cond, updates)
{
None => return Ok(()),
Some(builder) => builder.sql()?,
impl DbCursor {
/// To create the cursor instance.
pub fn new() -> Self {
DbCursor { offset: 0 }
impl Cursor for DbCursor {
async fn try_next(&mut self) -> Result<Option<Device>, Box<dyn StdError>> {
self.offset += 1;
Ok(None)
fn offset(&self) -> u64 {
self.offset
/// Transforms query conditions to the SQL builder.
fn build_where<'a>(builder: &'a mut SqlBuilder, cond: &QueryCond<'a>) -> &'a mut SqlBuilder {
if let Some(value) = cond.unit_id {
builder.and_where_eq("unit_id", quote(value));
if let Some(value) = cond.device_id {
builder.and_where_eq("device_id", quote(value));
if let Some(value) = cond.network_id {
builder.and_where_eq("network_id", quote(value));
if let Some(value) = cond.network_addrs {
let values: Vec<String> = value.iter().map(|&x| quote(x)).collect();
builder.and_where_in("network_addr", &values);
if let Some(value) = cond.device.as_ref() {
if let Some(unit_code) = value.unit_code {
builder.and_where_eq("unit_code", quote(unit_code));
builder.and_where_eq("unit_code", quote(""));
builder.and_where_eq("network_code", quote(value.network_code));
builder.and_where_eq("network_addr", quote(value.network_addr));
builder
fn build_list_where<'a>(
builder: &'a mut SqlBuilder,
cond: &ListQueryCond<'a>,
) -> &'a mut SqlBuilder {
if let Some(value) = cond.network_code {
builder.and_where_eq("network_code", quote(value));
if let Some(value) = cond.network_addr {
builder.and_where_eq("network_addr", quote(value));
} else if let Some(value) = cond.network_addrs {
if let Some(value) = cond.profile {
build_where_like(builder, "profile", value.to_lowercase().as_str());
if let Some(value) = cond.name_contains {
build_where_like(builder, "name", value.to_lowercase().as_str());
/// Transforms model options to the SQL builder.
fn build_limit_offset<'a>(builder: &'a mut SqlBuilder, opts: &ListOptions) -> &'a mut SqlBuilder {
if let Some(value) = opts.limit {
if value > 0 {
builder.limit(value);
if let Some(value) = opts.offset {
match opts.limit {
None => builder.limit(-1).offset(value),
Some(0) => builder.limit(-1).offset(value),
_ => builder.offset(value),
fn build_sort<'a>(builder: &'a mut SqlBuilder, opts: &ListOptions) -> &'a mut SqlBuilder {
if let Some(sort_cond) = opts.sort.as_ref() {
for cond in sort_cond.iter() {
let key = match cond.key {
SortKey::CreatedAt => "created_at",
SortKey::ModifiedAt => "modified_at",
SortKey::NetworkCode => "network_code",
SortKey::NetworkAddr => "network_addr",
SortKey::Profile => "profile",
SortKey::Name => "name",
builder.order_by(key, !cond.asc);
/// Transforms query conditions and the model object to the SQL builder.
fn build_update_where<'a>(
cond: &UpdateQueryCond<'a>,
) -> Option<&'a mut SqlBuilder> {
let mut count = 0;
if let Some((network_id, network_code)) = updates.network {
builder.set("network_id", quote(network_id));
builder.set("network_code", quote(network_code));
if let Some(value) = updates.network_addr {
builder.set("network_addr", quote(value));
if let Some(value) = updates.modified_at.as_ref() {
builder.set("modified_at", value.timestamp_millis());
if let Some(value) = updates.profile.as_ref() {
builder.set("profile", quote(value));
if let Some(value) = updates.name.as_ref() {
builder.set("name", quote(value));
if let Some(value) = updates.info {
match serde_json::to_string(value) {
Err(_) => {
builder.set("info", quote("{}"));
Ok(value) => {
builder.set("info", quote(value));
if count == 0 {
return None;
builder.and_where_eq("device_id", quote(cond.device_id));
Some(builder)