Batch text prediction with Gemini model using Google Cloud Storage

  import 
  
 time 
 from 
  
 google 
  
 import 
 genai 
 from 
  
 google.genai.types 
  
 import 
 CreateBatchJobConfig 
 , 
 JobState 
 , 
 HttpOptions 
 client 
 = 
 genai 
 . 
 Client 
 ( 
 http_options 
 = 
 HttpOptions 
 ( 
 api_version 
 = 
 "v1" 
 )) 
 # TODO(developer): Update and un-comment below line 
 # output_uri = "gs://your-bucket/your-prefix" 
 # See the documentation: https://googleapis.github.io/python-genai/genai.html#genai.batches.Batches.create 
 job 
 = 
 client 
 . 
 batches 
 . 
 create 
 ( 
 # To use a tuned model, set the model param to your tuned model using the following format: 
 # model="projects/{PROJECT_ID}/locations/{LOCATION}/models/{MODEL_ID} 
 model 
 = 
 "gemini-2.5-flash" 
 , 
 # Source link: https://storage.cloud.google.com/cloud-samples-data/batch/prompt_for_batch_gemini_predict.jsonl 
 src 
 = 
 "gs://cloud-samples-data/batch/prompt_for_batch_gemini_predict.jsonl" 
 , 
 config 
 = 
 CreateBatchJobConfig 
 ( 
 dest 
 = 
 output_uri 
 ), 
 ) 
 print 
 ( 
 f 
 "Job name: 
 { 
 job 
 . 
 name 
 } 
 " 
 ) 
 print 
 ( 
 f 
 "Job state: 
 { 
 job 
 . 
 state 
 } 
 " 
 ) 
 # Example response: 
 # Job name: projects/%PROJECT_ID%/locations/us-central1/batchPredictionJobs/9876453210000000000 
 # Job state: JOB_STATE_PENDING 
 # See the documentation: https://googleapis.github.io/python-genai/genai.html#genai.types.BatchJob 
 completed_states 
 = 
 { 
 JobState 
 . 
 JOB_STATE_SUCCEEDED 
 , 
 JobState 
 . 
 JOB_STATE_FAILED 
 , 
 JobState 
 . 
 JOB_STATE_CANCELLED 
 , 
 JobState 
 . 
 JOB_STATE_PAUSED 
 , 
 } 
 while 
 job 
 . 
 state 
 not 
 in 
 completed_states 
 : 
 time 
 . 
 sleep 
 ( 
 30 
 ) 
 job 
 = 
 client 
 . 
 batches 
 . 
 get 
 ( 
 name 
 = 
 job 
 . 
 name 
 ) 
 print 
 ( 
 f 
 "Job state: 
 { 
 job 
 . 
 state 
 } 
 " 
 ) 
 # Example response: 
 # Job state: JOB_STATE_PENDING 
 # Job state: JOB_STATE_RUNNING 
 # Job state: JOB_STATE_RUNNING 
 # ... 
 # Job state: JOB_STATE_SUCCEEDED