Python flask APIs
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
from sqlalchemy import and_
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///patient_data.db' # Use SQLite for simplicity
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
class PatientData(db.Model):
id = db.Column(db.Integer, primary_key=True)
patient_id = db.Column(db.String(50), nullable=False)
timestamp = db.Column(db.DateTime, nullable=False)
temperature = db.Column(db.Float)
blood_pressure = db.Column(db.String(20))
heart_rate = db.Column(db.Integer)
oxygen_saturation = db.Column(db.Integer)
ecg = db.Column(db.String(20))
ppg = db.Column(db.String(20))
def to_dict(self):
return {
"timestamp": self.timestamp.isoformat(),
"temperature": self.temperature,
"blood_pressure": self.blood_pressure,
"heart_rate": self.heart_rate,
"oxygen_saturation": self.oxygen_saturation,
"ecg": self.ecg,
"ppg": self.ppg
}
@app.route('/api/patient_data', methods=['GET'])
def get_patient_data():
patient_id = request.args.get('patient_id')
start_time = request.args.get('start_time')
end_time = request.args.get('end_time')
# Validate input parameters
if not all([patient_id, start_time, end_time]):
return jsonify({"error": "Missing required parameters"}), 400
try:
start_time = datetime.fromisoformat(start_time)
end_time = datetime.fromisoformat(end_time)
except ValueError:
return jsonify({"error": "Invalid datetime format. Use ISO format (YYYY-MM-DDTHH:MM:SS)"}), 400
if start_time > end_time:
return jsonify({"error": "Start time must be before end time"}), 400
# Query the database for patient data
patient_data = PatientData.query.filter(
and_(
PatientData.patient_id == patient_id,
PatientData.timestamp >= start_time,
PatientData.timestamp <= end_time
)
).order_by(PatientData.timestamp).all()
return jsonify({
"patient_id": patient_id,
"start_time": start_time.isoformat(),
"end_time": end_time.isoformat(),
"data": [data.to_dict() for data in patient_data]
})
@app.route('/api/add_patient_data', methods=['POST'])
def add_patient_data():
data = request.json
new_data = PatientData(
patient_id=data['patient_id'],
timestamp=datetime.fromisoformat(data['timestamp']),
temperature=data['temperature'],
blood_pressure=data['blood_pressure'],
heart_rate=data['heart_rate'],
oxygen_saturation=data['oxygen_saturation'],
ecg=data['ecg'],
ppg=data['ppg']
)
db.session.add(new_data)
db.session.commit()
return jsonify({"message": "Data added successfully"}), 201
if __name__ == '__main__':
with app.app_context():
db.create_all() # Create database tables
app.run(debug=True)
Comments
Post a Comment