AR

الوثائققواعد بيانات SQL

استخدم MySQL أو MariaDB مع Nexy

لاستخدام MySQL أو MariaDB مع Nexy، قم بتثبيت برنامج التشغيل pymysql، وقم بتعيين DATABASE_URL إلى عنوان URL mysql+pymysql:// أو mariadb+pymysql:// مع charset=utf8mb4، واستعلام عنه من خلال SQLModel.

قم بتشغيل MySQL أو MariaDB محليًا

docker run --name nexy-mysql -e MYSQL_ROOT_PASSWORD=secret -e MYSQL_DATABASE=app -p 3306:3306 -d mysql:8.4docker run --name nexy-mariadb -e MARIADB_ROOT_PASSWORD=secret -e MARIADB_DATABASE=app -p 3306:3306 -d mariadb:11

تثبيت برنامج التشغيل

uv add pymysql

pymysql طاهر Python ولا يحتاج إلى غيره. بالنسبة للكود غير المتزامن، يستخدمه aiomysql أدناه:

uv add aiomysql

سلسلة الاتصال

.env
DATABASE_URL=mysql+pymysql://root:secret@localhost:3306/app?charset=utf8mb4
عنوان URL الخادم والسائق
mysql+pymysql://user:pass@host:3306/db?charset=utf8mb4 ماي إس كيو إل، المزامنة
mariadb+pymysql://user:pass@host:3306/db?charset=utf8mb4 MariaDB، متزامنة مع SQL الخاصة بـ MariaDB
mysql+aiomysql://user:pass@host:3306/db?charset=utf8mb4 MySQL أو MariaDB، غير متزامن

الاحتفاظ بـ charset=utf8mb4: يقوم utf8 في MySQL بتخزين ثلاثة بايتات لكل حرف، ويحتاج الرمز التعبيري إلى أربعة بايتات.

المحرك

يقوم MySQL بإغلاق الاتصال الذي ظل خاملاً لمدة wait_timeout ثانية، أي ثماني ساعات افتراضيًا. pool_recycle يسقط الاتصالات قبل ذلك، و pool_pre_ping يلتقط الباقي:

src/configs/db.py
import os from dotenv import load_dotenvfrom sqlmodel import create_engine load_dotenv() engine = create_engine(    os.environ["DATABASE_URL"],    pool_size=5,    max_overflow=10,    pool_recycle=3600,    pool_pre_ping=True,)

ومن دونهم أول طلب الصبح يجيب MySQL server has gone away.

نموذج: إعطاء الأوتار طولاً

يحتاج MySQL إلى طول في كل VARCHAR. SQLModel يكتب VARCHAR(255) لـ str بلا شيء؛ اضبط max_length حيث تعرف الحجم، واستخدم Text للمحتوى الطويل:

src/models/article_model.py
from datetime import datetime from sqlalchemy import Column, DateTime, Text, funcfrom sqlmodel import Field, SQLModel  class Article(SQLModel, table=True):    id: int | None = Field(default=None, primary_key=True)    slug: str = Field(max_length=120, unique=True)    title: str = Field(max_length=200)    body: str = Field(sa_column=Column(Text))    created_at: datetime | None = Field(        default=None, sa_column=Column(DateTime, server_default=func.now())    )

يجب أن يتناسب VARCHAR الفريد أو المفهرس مع الفهرس: 3072 بايت مع InnoDB، أي ما يصل إلى 768 حرفًا في utf8mb4.

جلسة لكل طلب

src/configs/session.py
from collections.abc import Iterator from sqlmodel import Session from src.configs.db import engine  def get_session() -> Iterator[Session]:    with Session(engine) as session:        yield session

القراءة والكتابة من المعالج

src/routes/api/articles/index.py
from pydantic import BaseModelfrom sqlmodel import Session, select from nexy.http import Depends, JSONResponsefrom src.configs.session import get_sessionfrom src.models.article_model import Article  class NewArticle(BaseModel):    slug: str    title: str    body: str  async def GET(session: Session = Depends(get_session)):    return session.exec(select(Article).order_by(Article.id.desc()).limit(20)).all()  async def POST(body: NewArticle, session: Session = Depends(get_session)):    article = Article(**body.model_dump())    session.add(article)    session.commit()    session.refresh(article)    return JSONResponse(article.model_dump(mode="json"), status_code=201)
src/apps/articles/articles_api_controller.py
from pydantic import BaseModelfrom sqlmodel import Session, select from nexy.decorators import Controllerfrom nexy.http import Depends, JSONResponsefrom src.configs.session import get_sessionfrom src.models.article_model import Article  class NewArticle(BaseModel):    slug: str    title: str    body: str  @Controller("/api/articles")class ArticlesApiController:    async def get(self, session: Session = Depends(get_session)):        return session.exec(select(Article).order_by(Article.id.desc()).limit(20)).all()     async def post(self, body: NewArticle, session: Session = Depends(get_session)):        article = Article(**body.model_dump())        session.add(article)        session.commit()        session.refresh(article)        return JSONResponse(article.model_dump(mode="json"), status_code=201)

قم بإدراج وحدة التحكم في controllers من الوحدة الجذرية.

المفتاح المكرر هو 409

مقالة ثانية بنفس السلسلة الثابتة تكسر الفهرس الفريد. اكتشف الخطأ وأجب بالحالة:

src/routes/api/articles/[slug].py
from pydantic import BaseModelfrom sqlalchemy.exc import IntegrityErrorfrom sqlmodel import Session, select from nexy.http import Depends, HTTPExceptionfrom src.configs.session import get_sessionfrom src.models.article_model import Article  class Rename(BaseModel):    slug: str  async def PATCH(slug: str, body: Rename, session: Session = Depends(get_session)):    article = session.exec(select(Article).where(Article.slug == slug)).first()    if article is None:        raise HTTPException(status_code=404, detail="No such article")    article.slug = body.slug    try:        session.commit()    except IntegrityError:        session.rollback()        raise HTTPException(status_code=409, detail="Slug already taken")    return article
src/apps/articles/articles_by_slug_api_controller.py
from pydantic import BaseModelfrom sqlalchemy.exc import IntegrityErrorfrom sqlmodel import Session, select from nexy.decorators import Controllerfrom nexy.http import Depends, HTTPExceptionfrom src.configs.session import get_sessionfrom src.models.article_model import Article  class Rename(BaseModel):    slug: str  @Controller("/api/articles/{slug}")class ArticlesBySlugApiController:    async def patch(self, slug: str, body: Rename, session: Session = Depends(get_session)):        article = session.exec(select(Article).where(Article.slug == slug)).first()        if article is None:            raise HTTPException(status_code=404, detail="No such article")        article.slug = body.slug        try:            session.commit()        except IntegrityError:            session.rollback()            raise HTTPException(status_code=409, detail="Slug already taken")        return article

قم بإدراج وحدة التحكم في controllers من الوحدة الجذرية.

إقرأ من إحدى الصفحات

src/services/articles.py
from sqlmodel import Session, select from src.configs.db import enginefrom src.models.article_model import Article  def latest_articles() -> list[Article]:    with Session(engine) as session:        return list(session.exec(select(Article).order_by(Article.id.desc()).limit(5)))
src/routes/articles.nexy
---from src.services.articles import latest_articles articles = latest_articles()---<ul>  {% for article in articles %}  <li><a href="/articles/{{ article.slug | e }}">{{ article.title | e }}</a></li>  {% endfor %}</ul>
src/apps/articles/articles_view.nexy
---from src.services.articles import latest_articles articles = latest_articles()---<ul>  {% for article in articles %}  <li><a href="/articles/{{ article.slug | e }}">{{ article.title | e }}</a></li>  {% endfor %}</ul>
src/apps/articles/articles_controller.py
from nexy.decorators import Controllerfrom nexy.hooks import useViews  @Controller("/articles")class ArticlesController:    async def get(self):        return useViews("src/apps/articles/articles_view.nexy")

فقط الأسماء المعينة في المستوى الأعلى من المادة الأمامية تصل إلى القالب، وبالتالي فإن الاستعلام يعيش في وظيفة.

يزعج

يقوم MySQL وMariaDB بكتابة رسالة نصية كـ INSERT ... ON DUPLICATE KEY UPDATE:

src/services/counters.py
from sqlalchemy.dialects.mysql import insertfrom sqlmodel import Field, Session, SQLModel from src.configs.db import engine  class PageView(SQLModel, table=True):    path: str = Field(primary_key=True, max_length=255)    views: int = 0  def count_view(path: str) -> None:    statement = insert(PageView).values(path=path, views=1)    statement = statement.on_duplicate_key_update(views=PageView.views + 1)    with Session(engine) as session:        session.exec(statement)        session.commit()

البحث عن النص الكامل

قم بإضافة فهرس FULLTEXT في عملية الترحيل، ثم قم بالاستعلام عنه باستخدام MATCH ... AGAINST:

alembic/versions/xxxx_article_search.py
def upgrade() -> None:    op.execute("CREATE FULLTEXT INDEX article_search ON article (title, body)")
src/routes/api/search.py
from sqlalchemy import textfrom sqlmodel import Session, select from nexy.http import Dependsfrom src.configs.session import get_sessionfrom src.models.article_model import Article  async def GET(q: str, session: Session = Depends(get_session)):    match = text("MATCH (title, body) AGAINST (:q IN NATURAL LANGUAGE MODE)")    return session.exec(select(Article).where(match.bindparams(q=q))).all()
src/apps/search/search_api_controller.py
from sqlalchemy import textfrom sqlmodel import Session, select from nexy.decorators import Controllerfrom nexy.http import Dependsfrom src.configs.session import get_sessionfrom src.models.article_model import Article  @Controller("/api/search")class SearchApiController:    async def get(self, q: str, session: Session = Depends(get_session)):        match = text("MATCH (title, body) AGAINST (:q IN NATURAL LANGUAGE MODE)")        return session.exec(select(Article).where(match.bindparams(q=q))).all()

قم بإدراج وحدة التحكم في controllers من الوحدة الجذرية.

الهجرات

nx migrate --initnx migrate -m "create articles"nx migrate --upgrade

ينفذ MySQL كلاً من CREATE و ALTER من تلقاء نفسه، وبالتالي فإن الترحيل الذي يفشل في منتصف الطريق يظل نصف مطبق: أبقِ عمليات الترحيل صغيرة، مع تغيير واحد لكل منهما.

غير متزامن مع aiomysql

src/configs/async_db.py
import os from dotenv import load_dotenvfrom sqlalchemy.ext.asyncio import create_async_enginefrom sqlmodel.ext.asyncio.session import AsyncSession load_dotenv() async_engine = create_async_engine(    os.environ["DATABASE_URL"].replace("+pymysql", "+aiomysql", 1), pool_recycle=3600)  async def get_async_session():    async with AsyncSession(async_engine) as session:        yield session

MySQL أو MariaDB: ما الفرق

ماي إس كيو إل 8 ماريا دي بي 11
مخطط URL mysql+pymysql:// mariadb+pymysql://
JSON العمود النوع الأصلي اسم مستعار LONGTEXT بشيك
RETURNING لا نعم على INSERT و DELETE
تسلسلات لا CREATE SEQUENCE

رمز هذه الصفحة يعمل على كليهما.

إدارة MySQL

الخدمة ما يتغير
PlanetScale إضافة ?ssl_ca=/etc/ssl/certs/ca-certificates.crt؛ لا توجد مفاتيح خارجية بشكل افتراضي
أمازون آر دي إس، أورورا، كلاود إس كيو إل، أزور السماح بشبكة التطبيق واستخدام TLS

الخطوات التالية

حرِّر على GitHub

Enter يفتح النتيجة، وEsc يغلق.