title: FastAPI Sync SQLAlchemy to Async SQLAlchemy
body: <p>How can I turn this sync sqlalchemy query logic to async sqlalchemy.</p>
  <pre><code>def update_job_by_id(id:int, job: JobCreate,db: Session,owner_id):
      existing_job = db.query(Job).filter(Job.id == id)
      if not existing_job.first():
          return 0
      job.__dict__.update(owner_id=owner_id)  #update dictionary with new key value of owner_id
      existing_job.update(job.__dict__)
      db.commit()
      return 1
  </code></pre>
  <p>I tried to convert this logic into async sqlalchemy but no luck.Here is the async version of the above code:</p>
  <pre><code>async def update_job_by_id(self, id: int, job: PydanticValidationModel, owner_id):
          job_query = await self.db_session.get(db_table, id)
          if not job_query:
              return 0
          updated_job = update(db_table).where(db_table.id == id).values(job.__dict__.update(owner_id = owner_id)).execution_options(synchronize_session=&quot;fetch&quot;)
          await self.db_session.execute(updated_job)
          return 1
  </code></pre>
  <p>it produces this error:</p>
  <pre><code>AttributeError: 'NoneType' object has no attribute 'items'
  </code></pre>
  <p><strong>Scope:</strong></p>
  <p>Making a simple Job Board API using FastAPI and Async SQLAlchemy and I am struggling through update feature of the API in function <code>async def update_job_by_id</code> which I mention above first checking the <code>ID</code> of the job if <code>ID</code> is <code>True</code> then it will export the PydanticModel to dict object to update the PydanticModel with <code>owner_id</code> and finally updating the <code>Model</code> and update the job, but unfortunately when I try to update the <code>PydanticModel</code> <code>update(Job).values(**job_dict.update({&quot;owner_id&quot;: owner_id})).where(Job.id == id).execution_options(synchronize_session=&quot;fetch&quot;)</code> I get <code>None</code> instead of updated dict.</p>
  <p><strong>models/jobs.py</strong></p>
  <pre><code>from sqlalchemy import Column, Integer, String, Boolean, Date, ForeignKey
  from sqlalchemy.orm import relationship
  
  from db.base_class import Base
  
  
  
  class Job(Base):
      id = Column(Integer, primary_key=True, index=True)
      title = Column(String, nullable=False)
      company_name = Column(String, nullable=False)
      company_url = Column(String)
      location = Column(String, nullable=False)
      description = Column(String)
      date_posted = Column(Date)
      is_active = Column(Boolean, default=True)
      owner_id = Column(Integer, ForeignKey('user.id'))
      owner = relationship(&quot;User&quot;, back_populates=&quot;jobs&quot;)
  </code></pre>
  <p><strong>schemas/jobs.py</strong></p>
  <pre><code>from typing import Optional
  from pydantic import BaseModel
  from datetime import date, datetime
  
  
  class JobBase(BaseModel):
      title: Optional[str] = None
      company_name: Optional[str] = None
      company_url: Optional[str] = None
      location: Optional[str] = &quot;remote&quot;
      description: Optional[str] = None
      date_posted: Optional[date] = datetime.now().date()
  
  class JobCreate(JobBase):
      title: str
      company_name: str
      location: str
      description: str
  
  class ShowJob(JobBase):
      title: str
      company_name: str
      company_url: Optional[str]
      location: str
      date_posted: date
      description: str
  
      class Config():
          orm_mode = True
  </code></pre>
  <p><strong>routes/route_jobs.py</strong></p>
  <pre><code>from typing import List
  from fastapi import APIRouter, HTTPException, status
  from fastapi import Depends
  
  from db.repository.job_board_dal import job_board
  from schemas.jobs import JobCreate, ShowJob
  from db.repository.job_board_dal import Job
  from depends import get_db
  
  router = APIRouter()
  
  @router.post(&quot;/create-job&quot;,response_model=ShowJob)
  async def create_user(Job: JobCreate, jobs: Job = Depends(get_db)):
      owner_id = 1
      return await jobs.create_new_job(Job, owner_id)
  
  @router.get(&quot;/get/{id}&quot;, response_model=ShowJob)
  async def retrieve_job_by_id(id:int, id_job: job_board = Depends(get_db)):
      job_id = await job_board.retrieve_job(id_job, id=id)
      if not job_id:
          raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
          detail=f&quot;Job with id {id} does not exist&quot;)
      return job_id
  
  @router.get(&quot;/all&quot;, response_model=List[ShowJob])
  async def retrieve_all_jobs(all_jobs: job_board = Depends(get_db)):
      return await all_jobs.get_all_jobs()
  
  @router.put(&quot;/update/{id}&quot;)
  async def update_job(id: int, job: JobCreate, job_update: job_board = Depends(get_db)):
      current_user = 1
      response = await job_update.update_job_by_id(id = id, job = job, owner_id = current_user)
      if not response:
          raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
          detail=f&quot;Job with id {id} does not exist&quot;)
      return {&quot;response&quot;: &quot;Successfully updated the Job.&quot;}
  </code></pre>
  <p><strong>db/repository/job_board_dal.py</strong></p>
  <pre><code>from typing import List
  
  
  from sqlalchemy import update
  from sqlalchemy.engine import result
  from sqlalchemy.orm import Session
  from sqlalchemy.future import select
  
  from schemas.users import UserCreate
  from schemas.jobs import JobCreate
  from db.models.users import User
  from db.models.jobs import Job
  from core.hashing import Hasher
  
  
  
  class job_board():
      def __init__(self, db_session: Session):
          self.db_session = db_session
  
      async def register_user(self, user: UserCreate):
          new_user = User(username=user.username,
          email=user.email,
          hashed_password=Hasher.get_password_hash(user.password),
          is_active = False,
          is_superuser=False
          )
          self.db_session.add(new_user)
          await self.db_session.flush()
          return new_user
  
      
      async def create_new_job(self, job: JobCreate, owner_id: int):
          new_job = Job(**job.dict(), owner_id = owner_id)
          self.db_session.add(new_job)
          await self.db_session.flush()
          return new_job
  
      async def retrieve_job(self, id:int):
          item = await self.db_session.get(Job, id)
          return item
  
      async def get_all_jobs(self) -&gt; List[Job]:
          query = await self.db_session.execute(select(Job).order_by(Job.is_active == True))
          return query.scalars().all()
  
      async def update_job_by_id(self, id: int, job: JobCreate, owner_id):
          _job = await self.db_session.execute(select(Job).where(Job.id==id))
          (result, ) = _job.one()
          job_dict = job.dict()
          print(job_dict.update({&quot;owner_id&quot;: owner_id}))
          
          #print(job_update)
          if not result:
              return 0
          job_update = update(Job).values(job_dict.update({&quot;owner_id&quot;: owner_id})).where(Job.id == id).execution_options(synchronize_session=&quot;fetch&quot;)
          return await self.db_session.execute(job_update)
  </code></pre>
  <p>If anyone can point out what am I missing here it would be much appreciated.</p>