29 lines
877 B
Python
29 lines
877 B
Python
import sqlite3
|
|
import threading
|
|
from fastapi import FastAPI
|
|
|
|
class SQLiteManager:
|
|
def __init__(self, db_path: str):
|
|
self.db_path = db_path
|
|
self.connection = sqlite3.connect(db_path, check_same_thread=False)
|
|
self.connection.row_factory = sqlite3.Row
|
|
self.connection.execute("PRAGMA journal_mode=WAL")
|
|
self.lock = threading.RLock()
|
|
|
|
def query(self, sql: str, params: tuple = ()):
|
|
with self.lock:
|
|
cursor = self.connection.cursor()
|
|
cursor.execute(sql, params)
|
|
return cursor.fetchone()
|
|
|
|
db_manager = SQLiteManager("db.sqlite3")
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get("/cnes/{cnes_code}")
|
|
async def get_cnes(cnes_code: str):
|
|
result = db_manager.query(
|
|
"SELECT * FROM cnes_estabelecimentos WHERE CO_CNES = ? LIMIT 1",
|
|
(cnes_code,)
|
|
)
|
|
return dict(result) if result else None |