code: import os
  import pandas as pd
  from datetime import datetime
  import time
  import subprocess
  from docarray import DocumentArray, Document
  import json
  import pyarrow as pa
  import lancedb
  
  from google.cloud import bigquery
  
  
  GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "passculture-data-ehp")
  ENV_SHORT_NAME = os.environ.get("ENV_SHORT_NAME", "dev")
  BIGQUERY_CLEAN_DATASET = f"clean_{ENV_SHORT_NAME}"
  BIGQUERY_ANALYTICS_DATASET = f"analytics_{ENV_SHORT_NAME}"
  MODELS_RESULTS_TABLE_NAME = "mlflow_training_results"
  
  
  item_columns = [
      "vector",
      "item_id",
      "booking_number_desc",
      "booking_trend_desc",
      "booking_creation_trend_desc",
      "booking_release_trend_desc",
      "raw_embeddings",
      "topic_id",
      "cluster_id",
      "category",
      "subcategory_id",
      "search_group_name",
      "offer_type_label",
      "offer_type_domain",
      "gtl_id",
      "gtl_l1",
      "gtl_l2",
      "gtl_l3",
      "gtl_l4",
      "is_numerical",
      "is_national",
      "is_geolocated",
      "is_underage_recommendable",
      "is_restrained",
      "is_sensitive",
      "offer_is_duo",
      "booking_number",
      "booking_number_last_7_days",
      "booking_number_last_14_days",
      "booking_number_last_28_days",
      "stock_price",
      "offer_creation_date",
      "stock_beginning_date",
      "total_offers",
      "example_offer_id",
      "example_offer_name",
      "example_venue_id",
      "example_venue_latitude",
      "example_venue_longitude",
  ]
  
  
  def save_experiment(experiment_name, model_name, serving_container, run_id):
      log_results = {
          "execution_date": datetime.now().isoformat(),
          "experiment_name": experiment_name,
          "model_name": model_name,
          "model_type": "custom",
          "run_id": run_id,
          "run_start_time": int(time.time() * 1000.0),
          "run_end_time": int(time.time() * 1000.0),
          "artifact_uri": None,
          "serving_container": serving_container,
      }
  
      client = bigquery.Client()
      table_id = f"""{BIGQUERY_CLEAN_DATASET}.{MODELS_RESULTS_TABLE_NAME}"""
  
      job_config = bigquery.LoadJobConfig(
          schema=[
              bigquery.SchemaField("execution_date", "STRING"),
              bigquery.SchemaField("experiment_name", "STRING"),
              bigquery.SchemaField("model_name", "STRING"),
              bigquery.SchemaField("model_type", "STRING"),
              bigquery.SchemaField("run_id", "STRING"),
              bigquery.SchemaField("run_start_time", "INTEGER"),
              bigquery.SchemaField("run_end_time", "INTEGER"),
              bigquery.SchemaField("artifact_uri", "STRING"),
              bigquery.SchemaField("serving_container", "STRING"),
          ]
      )
      job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND
  
      df = pd.DataFrame.from_dict([log_results], orient="columns")
  
      job = client.load_table_from_dataframe(df, table_id, job_config=job_config)
      job.result()
  
  
  def deploy_container(serving_container, workers):
      command = f"sh ./deploy_to_docker_registery.sh {serving_container} {workers}"
      results = subprocess.Popen(
          command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
      )
      # TODO handle errors
      for line in results.stdout:
          print(line.rstrip().decode("utf-8"))
  
  
  def get_items_metadata():
      client = bigquery.Client()
  
      sql = f"""
          SELECT 
          *, 
          ROW_NUMBER() OVER (ORDER BY booking_number DESC) as booking_number_desc,
          ROW_NUMBER() OVER (ORDER BY booking_trend DESC) as booking_trend_desc,
          ROW_NUMBER() OVER (ORDER BY booking_creation_trend DESC) as booking_creation_trend_desc,
          ROW_NUMBER() OVER (ORDER BY booking_release_trend DESC) as booking_release_trend_desc
          FROM `{GCP_PROJECT_ID}.{BIGQUERY_ANALYTICS_DATASET}.recommendable_items_raw`
      """
      return client.query(sql).to_dataframe()
  
  
  def get_users_metadata():
      client = bigquery.Client()
  
      sql = f"""
          SELECT 
              user_id,
              user_total_deposit_amount,
              user_current_deposit_type,
              COALESCE(user_theoretical_remaining_credit, user_last_deposit_amount) as user_theoretical_remaining_credit
          FROM `{GCP_PROJECT_ID}.{BIGQUERY_ANALYTICS_DATASET}.enriched_user_data` 
      """
      return client.query(sql).to_dataframe()
  
  
  def to_ts(f):
      try:
          return float(f.timestamp())
      except:
          return 0.0
  
  
  def to_float(f):
      try:
          return float(f)
      except:
          return None
  
  
  def save_model_type(model_type):
      with open("./metadata/model_type.json", "w") as file:
          json.dump(model_type, file)
  
  
  def get_table_batches(item_embedding_dict: dict, items_df, emb_size):
      for row in items_df.itertuples():
          embedding_id = item_embedding_dict.get(row.item_id, None)
          if embedding_id is not None:
              _item_id = row.item_id
              yield pa.RecordBatch.from_arrays(
                  [
                      pa.array([embedding_id], pa.list_(pa.float32(), emb_size)),
                      pa.array([_item_id], pa.utf8()),
                      pa.array(
                          [[float(row.booking_number_desc)]], pa.list_(pa.float32(), 1)
                      ),
                      pa.array(
                          [[float(row.booking_trend_desc)]],
                          pa.list_(pa.float32(), 1),
                      ),
                      pa.array(
                          [[float(row.booking_creation_trend_desc)]],
                          pa.list_(pa.float32(), 1),
                      ),
                      pa.array(
                          [[float(row.booking_release_trend_desc)]],
                          pa.list_(pa.float32(), 1),
                      ),
                      pa.array([embedding_id], pa.list_(pa.float32(), emb_size)),
                      pa.array([str(row.topic_id or "")], pa.utf8()),
                      pa.array([str(row.cluster_id or "")], pa.utf8()),
                      pa.array([str(row.category or "")], pa.utf8()),
                      pa.array([str(row.subcategory_id or "")], pa.utf8()),
                      pa.array([str(row.search_group_name or "")], pa.utf8()),
                      pa.array([str(row.offer_type_label or "")], pa.utf8()),
                      pa.array([str(row.offer_type_domain or "")], pa.utf8()),
                      pa.array([str(row.gtl_id or "")], pa.utf8()),
                      pa.array([str(row.gtl_l1 or "")], pa.utf8()),
                      pa.array([str(row.gtl_l2 or "")], pa.utf8()),
                      pa.array([str(row.gtl_l3 or "")], pa.utf8()),
                      pa.array([str(row.gtl_l4 or "")], pa.utf8()),
                      pa.array([to_float(row.is_numerical)], pa.float32()),
                      pa.array([to_float(row.is_national)], pa.float32()),
                      pa.array([to_float(row.is_geolocated)], pa.float32()),
                      pa.array([to_float(row.is_underage_recommendable)], pa.float32()),
                      pa.array([to_float(row.is_restrained)], pa.float32()),
                      pa.array([to_float(row.is_sensitive)], pa.float32()),
                      pa.array([to_float(row.offer_is_duo)], pa.float32()),
                      pa.array([to_float(row.booking_number)], pa.float32()),
                      pa.array([to_float(row.booking_number_last_7_days)], pa.float32()),
                      pa.array([to_float(row.booking_number_last_14_days)], pa.float32()),
                      pa.array([to_float(row.booking_number_last_28_days)], pa.float32()),
                      pa.array([to_float(row.stock_price)], pa.float32()),
                      pa.array([to_ts(row.offer_creation_date)], pa.float32()),
                      pa.array([to_ts(row.stock_beginning_date)], pa.float32()),
                      # if unique
                      pa.array([to_float(row.total_offers)], pa.float32()),
                      pa.array([str(row.example_offer_id or "")], pa.utf8()),
                      pa.array([str(row.example_offer_name or "")], pa.utf8()),
                      pa.array([str(row.example_venue_id or "")], pa.utf8()),
                      pa.array(
                          [to_float(row.example_venue_latitude or 0.0)], pa.float32()
                      ),
                      pa.array(
                          [to_float(row.example_venue_longitude or 0.0)], pa.float32()
                      ),
                  ],
                  item_columns,
              )
  
  
  def create_items_table(
      item_embedding_dict, items_df, emb_size, uri="./metadata/vector"
  ):
      data = pa.Table.from_batches(
          get_table_batches(item_embedding_dict, items_df, emb_size)
      )
      db = lancedb.connect(uri)
      db.drop_database()
      table = db.create_table("items", data=data)
      table.create_index(num_partitions=1024, num_sub_vectors=32)
  
  
  def get_item_docs(item_embedding_dict, items_df):
      docs = DocumentArray()
      for row in items_df.itertuples():
          embedding_id = item_embedding_dict.get(row.item_id, None)
          if embedding_id is not None:
              _item_id = row.item_id
              docs.append(Document(id=str(_item_id), embedding=embedding_id))
  
      if len(docs) == 0:
          raise Exception("Item Document is empty. Does the model match the query ?")
  
      return docs
  
  
  def get_user_docs(user_dict):
      docs = DocumentArray()
      for k, v in user_dict.items():
          docs.append(Document(id=str(k), embedding=v))
      return docs
  
apis: lancedb.connect
extract_api: [((237, 293), 'os.environ.get', 'os.environ.get', (['"""GCP_PROJECT_ID"""', '"""passculture-data-ehp"""'], {}), "('GCP_PROJECT_ID', 'passculture-data-ehp')\n", (251, 293), False, 'import os\n'), ((311, 350), 'os.environ.get', 'os.environ.get', (['"""ENV_SHORT_NAME"""', '"""dev"""'], {}), "('ENV_SHORT_NAME', 'dev')\n", (325, 350), False, 'import os\n'), ((1925, 1942), 'google.cloud.bigquery.Client', 'bigquery.Client', ([], {}), '()\n', (1940, 1942), False, 'from google.cloud import bigquery\n'), ((2721, 2776), 'pandas.DataFrame.from_dict', 'pd.DataFrame.from_dict', (['[log_results]'], {'orient': '"""columns"""'}), "([log_results], orient='columns')\n", (2743, 2776), True, 'import pandas as pd\n'), ((3023, 3115), 'subprocess.Popen', 'subprocess.Popen', (['command'], {'shell': '(True)', 'stdout': 'subprocess.PIPE', 'stderr': 'subprocess.STDOUT'}), '(command, shell=True, stdout=subprocess.PIPE, stderr=\n    subprocess.STDOUT)\n', (3039, 3115), False, 'import subprocess\n'), ((3268, 3285), 'google.cloud.bigquery.Client', 'bigquery.Client', ([], {}), '()\n', (3283, 3285), False, 'from google.cloud import bigquery\n'), ((3859, 3876), 'google.cloud.bigquery.Client', 'bigquery.Client', ([], {}), '()\n', (3874, 3876), False, 'from google.cloud import bigquery\n'), ((8567, 8587), 'lancedb.connect', 'lancedb.connect', (['uri'], {}), '(uri)\n', (8582, 8587), False, 'import lancedb\n'), ((8786, 8801), 'docarray.DocumentArray', 'DocumentArray', ([], {}), '()\n', (8799, 8801), False, 'from docarray import DocumentArray, Document\n'), ((9222, 9237), 'docarray.DocumentArray', 'DocumentArray', ([], {}), '()\n', (9235, 9237), False, 'from docarray import DocumentArray, Document\n'), ((4537, 4564), 'json.dump', 'json.dump', (['model_type', 'file'], {}), '(model_type, file)\n', (4546, 4564), False, 'import json\n'), ((1559, 1573), 'datetime.datetime.now', 'datetime.now', ([], {}), '()\n', (1571, 1573), False, 'from datetime import datetime\n'), ((1753, 1764), 'time.time', 'time.time', ([], {}), '()\n', (1762, 1764), False, 'import time\n'), ((1804, 1815), 'time.time', 'time.time', ([], {}), '()\n', (1813, 1815), False, 'import time\n'), ((2089, 2137), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""execution_date"""', '"""STRING"""'], {}), "('execution_date', 'STRING')\n", (2109, 2137), False, 'from google.cloud import bigquery\n'), ((2151, 2200), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""experiment_name"""', '"""STRING"""'], {}), "('experiment_name', 'STRING')\n", (2171, 2200), False, 'from google.cloud import bigquery\n'), ((2214, 2258), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""model_name"""', '"""STRING"""'], {}), "('model_name', 'STRING')\n", (2234, 2258), False, 'from google.cloud import bigquery\n'), ((2272, 2316), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""model_type"""', '"""STRING"""'], {}), "('model_type', 'STRING')\n", (2292, 2316), False, 'from google.cloud import bigquery\n'), ((2330, 2370), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""run_id"""', '"""STRING"""'], {}), "('run_id', 'STRING')\n", (2350, 2370), False, 'from google.cloud import bigquery\n'), ((2384, 2433), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""run_start_time"""', '"""INTEGER"""'], {}), "('run_start_time', 'INTEGER')\n", (2404, 2433), False, 'from google.cloud import bigquery\n'), ((2447, 2494), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""run_end_time"""', '"""INTEGER"""'], {}), "('run_end_time', 'INTEGER')\n", (2467, 2494), False, 'from google.cloud import bigquery\n'), ((2508, 2554), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""artifact_uri"""', '"""STRING"""'], {}), "('artifact_uri', 'STRING')\n", (2528, 2554), False, 'from google.cloud import bigquery\n'), ((2568, 2619), 'google.cloud.bigquery.SchemaField', 'bigquery.SchemaField', (['"""serving_container"""', '"""STRING"""'], {}), "('serving_container', 'STRING')\n", (2588, 2619), False, 'from google.cloud import bigquery\n'), ((4998, 5007), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (5005, 5007), True, 'import pyarrow as pa\n'), ((5791, 5800), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (5798, 5800), True, 'import pyarrow as pa\n'), ((5861, 5870), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (5868, 5870), True, 'import pyarrow as pa\n'), ((5929, 5938), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (5936, 5938), True, 'import pyarrow as pa\n'), ((6003, 6012), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6010, 6012), True, 'import pyarrow as pa\n'), ((6080, 6089), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6087, 6089), True, 'import pyarrow as pa\n'), ((6156, 6165), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6163, 6165), True, 'import pyarrow as pa\n'), ((6233, 6242), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6240, 6242), True, 'import pyarrow as pa\n'), ((6299, 6308), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6306, 6308), True, 'import pyarrow as pa\n'), ((6365, 6374), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6372, 6374), True, 'import pyarrow as pa\n'), ((6431, 6440), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6438, 6440), True, 'import pyarrow as pa\n'), ((6497, 6506), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6504, 6506), True, 'import pyarrow as pa\n'), ((6563, 6572), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (6570, 6572), True, 'import pyarrow as pa\n'), ((6634, 6646), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (6644, 6646), True, 'import pyarrow as pa\n'), ((6707, 6719), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (6717, 6719), True, 'import pyarrow as pa\n'), ((6782, 6794), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (6792, 6794), True, 'import pyarrow as pa\n'), ((6869, 6881), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (6879, 6881), True, 'import pyarrow as pa\n'), ((6944, 6956), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (6954, 6956), True, 'import pyarrow as pa\n'), ((7018, 7030), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7028, 7030), True, 'import pyarrow as pa\n'), ((7092, 7104), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7102, 7104), True, 'import pyarrow as pa\n'), ((7168, 7180), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7178, 7180), True, 'import pyarrow as pa\n'), ((7256, 7268), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7266, 7268), True, 'import pyarrow as pa\n'), ((7345, 7357), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7355, 7357), True, 'import pyarrow as pa\n'), ((7434, 7446), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7444, 7446), True, 'import pyarrow as pa\n'), ((7507, 7519), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7517, 7519), True, 'import pyarrow as pa\n'), ((7585, 7597), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7595, 7597), True, 'import pyarrow as pa\n'), ((7664, 7676), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7674, 7676), True, 'import pyarrow as pa\n'), ((7770, 7782), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (7780, 7782), True, 'import pyarrow as pa\n'), ((7849, 7858), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (7856, 7858), True, 'import pyarrow as pa\n'), ((7927, 7936), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (7934, 7936), True, 'import pyarrow as pa\n'), ((8003, 8012), 'pyarrow.utf8', 'pa.utf8', ([], {}), '()\n', (8010, 8012), True, 'import pyarrow as pa\n'), ((8116, 8128), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (8126, 8128), True, 'import pyarrow as pa\n'), ((8254, 8266), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (8264, 8266), True, 'import pyarrow as pa\n'), ((4931, 4943), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (4941, 4943), True, 'import pyarrow as pa\n'), ((5109, 5121), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (5119, 5121), True, 'import pyarrow as pa\n'), ((5271, 5283), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (5281, 5283), True, 'import pyarrow as pa\n'), ((5443, 5455), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (5453, 5455), True, 'import pyarrow as pa\n'), ((5614, 5626), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (5624, 5626), True, 'import pyarrow as pa\n'), ((5709, 5721), 'pyarrow.float32', 'pa.float32', ([], {}), '()\n', (5719, 5721), True, 'import pyarrow as pa\n')]