DBへCSVをインポートするツールを作っている その13 - Dieselのセットアップその4

引き続き、Rustから読み込む部分を作っていく。まずは、 src/lib.rs をサンプルから拝借してくる。なお、Dockerコンテナ内で動かすので、今回はdotenvを使用しない。

// src/lib.rs

// https://diesel.rs/guides/getting-started

pub mod schema;
pub mod models;

#[macro_use]
extern crate diesel;
// extern crate dotenv;

use diesel::prelude::*;
use diesel::pg::PgConnection;
// use dotenv::dotenv;
use std::env;

pub fn establish_connection() -> PgConnection {
    // dotenv().ok();

    let database_url = env::var("DATABASE_URL")
        .expect("DATABASE_URL must be set");
    PgConnection::establish(&database_url)
        .expect(&format!("Error connecting to {}", database_url))
}

続いて、 src/models.rs を作る。

// src/models.rs

#[derive(Queryable)]
pub struct Stock {
    pub code: i32,
    pub datetime: String,
    pub volume: i32,
    pub price: f32,
}

最後に src/schema.rs だが、これはマイグレーションの時に自動生成されるので、手動で手を加えない方が良いだろう。

// src/schema.rs

table! {
    stocks (code, datetime) {
        code -> Int4,
        datetime -> Timestamp,
        volume -> Int4,
        price -> Numeric,
    }
}

ここまでできたら、 src/main.rs を下記のように変更して、これからDBを読み出す準備をする。

// src/main.rs

extern crate time_and_sales_deliver;

use std::{sync::mpsc, thread};

use futures::executor;

use actix_web::{get, middleware, web, App, Responder, HttpResponse, HttpServer};

use self::time_and_sales_deliver::establish_connection;

#[get("/test/{datetime}")]
async fn test(web::Path(datetime): web::Path<String>) -> HttpResponse {
    // let sql = format!("ORDER BY abs(TIMESTAMPDIFF(second, datetime, \"{}\")) LIMIT 1", datetime);
    // println!("{}", sql);
    let connection = establish_connection();
    HttpResponse::NoContent().finish()
}

#[get("/{code}/{datetime}/index.html")]
async fn index(web::Path((code, datetime)): web::Path<(String, String)>) -> impl Responder {
    //read(datetime.as_str());
    format!("{{\"code\":\"{}\", \"datetime\":\"{}\"}}", code, datetime)
}

// https://github.com/actix/examples/pull/240/files
#[get("/stop")]
async fn stop(stopper: web::Data<mpsc::Sender<()>>) -> HttpResponse {
    // make request that sends message through the Sender
    stopper.send(()).unwrap();

    HttpResponse::NoContent().finish()
}

#[get("/hello")]
async fn hello() -> &'static str {
    "Hello world!"
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    std::env::set_var("RUST_LOG", "actix_server=debug,actix_web=debug");
    env_logger::init();

    // create a channel
    let (tx, rx) = mpsc::channel::<()>();

    let bind = "0.0.0.0:8080";

    // start server as normal but don't .await after .run() yet
    let server = HttpServer::new(move || {
        // give the server a Sender in .data
        let stopper = tx.clone();

        App::new()
            .data(stopper)
            .wrap(middleware::Logger::default())
            .service(hello)
            .service(index)
            .service(test)
            .service(stop)
    })
    .bind(&bind)?
    .run();

    // clone the Server handle
    let srv = server.clone();
    thread::spawn(move || {
        // wait for shutdown signal
        rx.recv().unwrap();

        // stop server gracefully
        executor::block_on(srv.stop(true))
    });

    // run server
    server.await
}

次回以降は、 src/lib.rs を編集して、読み出しを行う関数を作成する。