ورقة مرجعية سريعة عن SQLAlchemy
الإعداد
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///chat.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
تعريف النماذج
class Message(db.Model):
__tablename__ = 'messages'
id = db.Column(db.Integer, primary_key=True)
content = db.Column(db.String(500), nullable=False)
sender = db.Column(db.String(20), nullable=False)
timestamp = db.Column(db.DateTime, default=datetime.utcnow)
أنواع الأعمدة
| النوع |
النوع في Python |
الاستخدام |
db.Integer |
int |
المعرّفات والأعداد |
db.String(n) |
str |
النص القصير |
db.Text |
str |
النص الطويل |
db.Boolean |
bool |
القيم المنطقية |
db.DateTime |
datetime |
الطوابع الزمنية |
db.Float |
float |
الأرقام |
db.JSON |
dict/list |
البيانات المعقدة |
خيارات الأعمدة
db.Column(db.Integer, primary_key=True) # معرّف يتزايد تلقائيًا
db.Column(db.String, nullable=False) # حقل إلزامي
db.Column(db.String, unique=True) # لا تكرار للقيم
db.Column(db.String, default='user') # القيمة الافتراضية
db.Column(db.DateTime, default=datetime.utcnow) # طابع زمني تلقائي
إنشاء الجداول
# في صدفة Python أو عند بدء التشغيل
with app.app_context():
db.create_all()
إنشاء السجلات
msg = Message(content="Hello!", sender="user")
db.session.add(msg)
db.session.commit()
print(msg.id) # يُسند تلقائيًا
الاستعلام عن السجلات
# جميع السجلات
Message.query.all()
# تصفية
Message.query.filter_by(sender='user').all()
Message.query.filter(Message.content.contains('hello')).all()
# جلب حسب المعرّف
Message.query.get(5)
# الترتيب
Message.query.order_by(Message.timestamp.desc()).all()
Message.query.order_by(Message.timestamp.asc()).all()
# تحديد العدد
Message.query.limit(10).all()
# العد
Message.query.count()
Message.query.filter_by(sender='ai').count()
# أول نتيجة مطابقة
Message.query.first()
Message.query.filter_by(sender='user').first()
# تسلسل الاستدعاءات
Message.query.filter_by(sender='user').order_by(Message.timestamp.desc()).limit(5).all()
تحديث السجلات
msg = Message.query.get(1)
msg.content = "Updated text"
db.session.commit()
حذف السجلات
msg = Message.query.get(1)
db.session.delete(msg)
db.session.commit()
# حذف السجلات المطابقة
Message.query.filter_by(sender='user').delete()
db.session.commit()
التحويل إلى قاموس (من أجل JSON)
def to_dict(self):
return {
'id': self.id,
'content': self.content,
'sender': self.sender,
'timestamp': self.timestamp.isoformat() if self.timestamp else None
}