Tengo editor y article . Muchos editores pueden estar relacionados con muchos artículos y muchos artículos pueden tener muchos editores al mismo tiempo.
Mis tablas DB son
| identificación | tema | texto |
|---|---|---|
| 1 | Vacaciones de Año Nuevo | En este año... etc etc etc |
| identificación | nombre | |
|---|---|---|
| 1 | John Smith | algunos@correo electrónico |
| editor_id | ID del artículo |
|---|---|
| 1 | 1 |
mis modelos son
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey from sqlalchemy.orm import relationship from database import Base class Editor(Base): __tablename__ = "editor" id = Column(Integer, primary_key=True, index=True) name = Column(String(32), unique=False, index=False, nullable=True) email = Column(String(115), unique=True, index=True) articles = relationship("Article", secondary=EditorArticleRelation, back_populates="articles", cascade="all, delete") class Article(Base): __tablename__ = "article" id = Column(Integer, primary_key=True, index=True) subject = Column(String(32), unique=True, index=False) text = Column(String(256), unique=True, index=True, nullable=True) editors = relationship("Editor", secondary=EditorArticleRelation, back_populates="editors", cascade="all, delete") EditorArticleRelation = Table('editorarticlerelation', Base.metadata, Column('editor_id', Integer, ForeignKey('editor.id')), Column('article_id', Integer, ForeignKey('article.id')) )mis esquemas son
from typing import Optional, List from pydantic import BaseModel class EditorBase(BaseModel): name: Optional[str] email: str class EditorCreate(EditorBase): pass class Editor(EditorBase): id: int class Config: orm_mode = True class ArticleBase(BaseModel): subject: str text: str class ArticleCreate(ArticleBase): # WHAT I NEED TO SET HERE??? editor_ids: List[int] = [] class Article(ArticleBase): id: int editors: List[Editor] = [] class Config: orm_mode = Truemi basura
def create_article(db: Session, article_data: schema.ArticleCreate): db_article = model.Article(subject=article_data.subject, text=article_data.text, ??? HOW TO SET EDITORS HERE ???) db.add(db_article) db.commit() db.refresh(db_article) return db_articlemi ruta
@app.post("/articles/", response_model=schema.Article) def create_article(article_data: schema.ArticleCreate, db: Session = Depends(get_db)): db_article = crud.get_article_by_name(db, name=article_data.name) if db_article: raise HTTPException(status_code=400, detail="article already registered") if len(getattr(article_data, 'editor_ids', [])) > 0: ??? WHAT I NEED TO SET HERE??? return crud.create_article(db=db, article_data=article_data)Quiero publicar datos para la API de creación de artículos y resolver y agregar automáticamente relaciones de editor, o generar un error si alguno de los editores no existe:
{ "subject": "Fresh news" "text": "Today is ..." "editor_ids": [1, 2, ...] }HOW TO SET EDITORS HERE )?WHAT I NEED TO SET HERE )?WHAT I NEED TO SET HERE )?pydantic y sqlalchemy , cualquier información será bienvenida.No estoy seguro si mi solución es más efectiva, pero lo hice de esta manera:
... @app.post("/articles/", response_model=schema.Article) def create_article(article_data: schema.ArticleCreate, db: Session = Depends(get_db)): db_article = crud.get_article_by_name(db, name=article_data.name) if db_article: raise HTTPException(status_code=400, detail="article already registered") return crud.create_article(db=db, article_data=article_data) ... ... class ArticleCreate(ArticleBase): editor_ids: List[int] = [] ... def create_article(db: Session, article_data: schema.ArticleCreate): db_article = model.Article(subject=article_data.subject, text=article_data.text) if (editors := db.query(model.Editor).filter(model.Editor.id.in_(article_data.editor_ids))).count() == len(endpoint_data.topic_ids): db_article.topics.extend(editors) else: # even if at least one editor is not found, an error is raised # if existence is not matter you can skip this check and add relations only for existing data raise HTTPException(status_code=404, detail="editor not found") db.add(db_article) db.commit() db.refresh(db_article) return db_articleCualquier idea mejor es bienvenida
Una solución que encontré.
def create_user_groups(db: Session, user_groups: schemas.UserGroupsBase): db_user = db.query(models.User).filter(models.User.id == user_groups.id_user).first() db_group = db.query(models.Group).filter(models.Group.id == user_groups.id_group).first() if not db_user and db_group: raise HTTPException(status_code=409, detail="User or Group not found in system.") db_user.groups.append(db_group) db.add(db_user) db.commit() db.refresh(db_user) return db_user