class_docstr: A table for DAGs code.
  
  dag_code table contains code of DAG files synchronized by scheduler.
  
  For details on dag serialization see SerializedDagModel
class_docstr_tok: ["A","table","for","DAGs","code",".","dag_code","table","contains","code","of","DAG","files","synchronized","by","scheduler",".","For","details","on","dag","serialization","see","SerializedDagModel"]
docstr: Write code in bulk into database.
  
  :param filelocs: file paths of DAGs to sync
  :param session: ORM Session
docstr_tok: ["Write","code","in","bulk","into","database",".",":","param","filelocs",":","file","paths","of","DAGs","to","sync",":","param","session",":","ORM","Session"]
code: def bulk_sync_to_db(
      cls, filelocs: Iterable[str], session: Session = NEW_SESSION
  ) -> None:
      """Write code in bulk into database.
  
      :param filelocs: file paths of DAGs to sync
      :param session: ORM Session
      """
      filelocs = set(filelocs)
      filelocs_to_hashes = {
          fileloc: DagCode.dag_fileloc_hash(fileloc)
          for fileloc in filelocs
      }
      existing_orm_dag_codes = session.scalars(
          select(DagCode)
          .filter(DagCode.fileloc_hash.in_(filelocs_to_hashes.values()))
          .with_for_update(of=DagCode)
      ).all()
  
      if existing_orm_dag_codes:
          existing_orm_dag_codes_map = {
              orm_dag_code.fileloc: orm_dag_code
              for orm_dag_code in existing_orm_dag_codes
          }
      else:
          existing_orm_dag_codes_map = {}
  
      existing_orm_dag_codes_by_fileloc_hashes = {
          orm.fileloc_hash: orm for orm in existing_orm_dag_codes
      }
      existing_orm_filelocs = {
          orm.fileloc
          for orm in existing_orm_dag_codes_by_fileloc_hashes.values()
      }
      if not existing_orm_filelocs.issubset(filelocs):
          conflicting_filelocs = existing_orm_filelocs.difference(
              filelocs
          )
          hashes_to_filelocs = {
              DagCode.dag_fileloc_hash(fileloc): fileloc
              for fileloc in filelocs
          }
          message = ""
          for fileloc in conflicting_filelocs:
              filename = hashes_to_filelocs[
                  DagCode.dag_fileloc_hash(fileloc)
              ]
              message += (
                  f"Filename '{filename}' causes a hash collision in the "
                  f"database with '{fileloc}'. Please rename the file."
              )
          raise AirflowException(message)
  
      existing_filelocs = {
          dag_code.fileloc for dag_code in existing_orm_dag_codes
      }
      missing_filelocs = filelocs.difference(existing_filelocs)
  
      for fileloc in missing_filelocs:
          orm_dag_code = DagCode(
              fileloc, cls._get_code_from_file(fileloc)
          )
          session.add(orm_dag_code)
  
      for fileloc in existing_filelocs:
          current_version = existing_orm_dag_codes_by_fileloc_hashes[
              filelocs_to_hashes[fileloc]
          ]
          file_mod_time = datetime.fromtimestamp(
              os.path.getmtime(correct_maybe_zipped(fileloc)),
              tz=timezone.utc,
          )
  
          if file_mod_time > current_version.last_updated:
              orm_dag_code = existing_orm_dag_codes_map[fileloc]
              orm_dag_code.last_updated = file_mod_time
              orm_dag_code.source_code = cls._get_code_from_file(
                  orm_dag_code.fileloc
              )
              session.merge(orm_dag_code)
  
imports: from __future__ import annotations
  import logging
  import os
  import struct
  from datetime import datetime
  from typing import TYPE_CHECKING, Collection, Iterable
  from sqlalchemy import BigInteger, Column, String, Text, delete, select
  from sqlalchemy.dialects.mysql import MEDIUMTEXT
  from sqlalchemy.sql.expression import literal
  from airflow.exceptions import AirflowException, DagCodeNotFound
  from airflow.models.base import Base
  from airflow.utils import timezone
  from airflow.utils.file import correct_maybe_zipped, open_maybe_zipped
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.sqlalchemy import UtcDateTime
label_desc: Use image node_id 3 for calling the DagCode obj's underlying member method code with example usage: obj.bulk_sync_to_db(cls, filelocs, session) without return types