Практически любой проект состоит из базы данных и информации содержащейся в ней. Вся ценность информации в ее полезности и актуальности. Для актуализации информации ее а) для начала нужно добавить, б) периодически нужно актуализировать. Об этом мы сегодня и поговорим но в контексте определенного стека а именно SqlAlchemy, FastApi, Pydantic.

Итак допустим у нас есть таблица в базе данных Postgresql.

import uuid

from sqlalchemy import Column, String, Boolean, Integer, ARRAY, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.types import Date
from sqlalchemy.orm import relationship
from app.database.base import Base
from app.models.mixins import TimestampMixin


class User(TimestampMixin, Base):
    __tablename__ = "users"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
    first_name = Column(String, index=True)
    last_name = Column(String, nullable=False, index=True, comment="Фамилия") 
    patronymic = Column(String, nullable=True, index=True, comment="Отчество")
    email = Column(String, unique=True, index=True, nullable=False)
    phone = Column(String, unique=True, index=True, nullable=False)

А также есть соответствующая Pydantic схема для определения полей. Схемы у меня асбтрактные бездумно копироать не стоит привел для построения целостной картины.

from pydantic import BaseModel, EmailStr, UUID4, Field, validator

class UserBase(BaseModel):
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    patronymic: Optional[str] = None
    email: Optional[EmailStr] = None
    phone: Optional[str] = None

# Properties to receive via API on creation
class UserCreate(UserBase):
    email: EmailStr
    first_name: str
    last_name: str 
    password: str

class UserOut(BaseModel):
    id: Optional[UUID4]
    clinic_id: Optional[UUID4]
    clinic: Optional[ClinicResponse]
    email: Optional[EmailStr]
    first_name: Optional[str]
    last_name: Optional[str]
    patronymic: Optional[str]

    class Config:
        orm_mode = True

Далее у нас идет сам роут по которому мы будем получать запросы.

import os, random
from typing import List
import uuid
from uuid import uuid4
from fastapi import APIRouter, HTTPException, status, Depends, UploadFile, File
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi_pagination import Page
from fastapi_pagination.ext.async_sqlalchemy import paginate

from app.api.deps import get_current_active_user, get_session
from app.schemas.user import UserUpdate, UserAuth, UserOut, \
    UserListOut, UserCreate, PersonCreate, PersonOut
from app.core import security
from app import crud
from app.crud.crud_user import create_user as create_person, update_user as update_user_db
from app.core.config import settings
from app.S3_utils import get_s3_client, upload_file_to_bucket
from app.models.user import User as UserModel

router = APIRouter(prefix="/users", tags=["Аккаунты пользователей"])

@router.post("/update", summary="Обновление данных пользователя", response_model=PersonOut)
async def update_user(
    db: AsyncSession = Depends(get_session),
    current_user: UserOut = Depends(get_current_active_user),
    user_in: UserUpdate = None
    )-> PersonOut:
    user = await update_user_db(db=db, user_in=user_in)
    if user is None:
        raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail="Ошибка обновления данных"
    )
    return user

Ну и последний самый главный наш блок непосрредственно crud_user.py файл в котором и будет происходить работа с базой посредством orm — sqlalchemy, прекрасной библиотекой мощью которой я не перестаю удивляться.

import logging
from sqlalchemy import select, column
from typing import Any, Dict, Optional, Union
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from fastapi.encoders import jsonable_encoder
from pydantic import UUID4

from app.crud.base import CRUDBase
from app.models.user import User
from app.models.clinic import Clinic
from app.schemas.user import UserCreate, UserUpdate, PersonCreate
from app.core import security
from fastapi_pagination.ext.async_sqlalchemy import paginate

async def update_user(db: AsyncSession, user_in: UserUpdate):
    stmt = (
        update(User).
        where(User.id == user_in.id).
        values(user_in.dict(exclude_unset=True)).
        returning(User)
    )
    result = await db.execute(stmt)
    await db.commit()
    # logging.info(result.first())
    return result.first()

Вот так вот просто и эффективно можно обновить запись в базе при помощи sqlalchemy, причем обратите внимание в моем примере асинхронная sqlalchemy в случае обычных синхронных сессий все будет тоже самое, только не будет async/await. Ну и еще бы хотел обратить вниманием на строчку

values(user_in.dict(exclude_unset=True)).

Тут мы модель Pydantic передаем непосредственно в sqllachemy таким образом полностью раскрываю мощь Pydantic по валидации данных.