import sqlite3 from app.types import Order def connect(connection_string): connection = sqlite3.connect( connection_string, detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES ) connection.execute("pragma journal_mode=wal") return connection def apply_database_schema(connection): connection.execute( """ CREATE TABLE IF NOT EXISTS orders ( id integer primary key AUTOINCREMENT, side varchar, type varchar, instrument char(12), limit_price varchar, quantity int, created_at timestamp, status varchar DEFAULT('pending') ) """ ) def insert_order(connection, order: Order): result = connection.execute( """ INSERT INTO orders ( side, type, instrument, limit_price, quantity, created_at ) VALUES ( :side, :type, :instrument, :limit_price, :quantity, datetime() ) returning id, created_at """, { "side": order.side.value, "type": order.type_.value, "instrument": order.instrument, "limit_price": str(order.limit_price), "quantity": str(order.quantity), }, ) returned_row = result.fetchone() order.id_ = returned_row[0] order.created_at = returned_row[1] return order def mark_order_as(connection, order_id, status): connection.execute( """ UPDATE orders set status = :status where id = :order_id """, {"status": status, "order_id": order_id}, )