google_cloud_storage_transfer_one_time_job_between_buckets.py

1import datetime
2
3from google.cloud import storage_transfer
4
5
6def create_transfer_job(
7    project_id: str,
8    description: str,
9    source_bucket: str,
10    sink_bucket: str,
11    schedule_start_date: datetime.date,
12):
13    """Creates a one-time transfer job."""
14
15    client = storage_transfer.StorageTransferServiceClient()
16
17    # The identification of the Google Cloud project that owns the job
18    # project_id = 'my-project-id'
19
20    # A description of the job
21    # description = 'My transfer job'
22
23    # Google Cloud Storage source bucket name
24    # source_bucket = 'my-source-bucket'
25
26    # Google Cloud Storage destination bucket name
27    # sink_bucket = 'my-sink-bucket'
28
29    # The date to start the transfer
30    # schedule_start_date = datetime.date.today()
31
32    transfer_job_request = storage_transfer.CreateTransferJobRequest(
33        transfer_job=storage_transfer.TransferJob(
34            project_id=project_id,
35            description=description,
36            status=storage_transfer.TransferJob.Status.ENABLED,
37            transfer_spec=storage_transfer.TransferSpec(
38                gcs_data_source=storage_transfer.GcsData(bucket_name=source_bucket),
39                gcs_data_sink=storage_transfer.GcsData(bucket_name=sink_bucket),
40            ),
41            schedule=storage_transfer.Schedule(
42                schedule_start_date=storage_transfer.Date(
43                    year=schedule_start_date.year,
44                    month=schedule_start_date.month,
45                    day=schedule_start_date.day,
46                ),
47                schedule_end_date=storage_transfer.Date(
48                    year=schedule_start_date.year,
49                    month=schedule_start_date.month,
50                    day=schedule_start_date.day,
51                ),
52            ),
53        )
54    )
55
56    result = client.create_transfer_job(request=transfer_job_request)
57    print(f"Created transfer_job: {result.name}")
58
59
60if __name__ == "__main__":
61    # Example usage:
62    # create_transfer_job(
63    #     project_id='your-project-id',
64    #     description='Example transfer job',
65    #     source_bucket='your-source-bucket',
66    #     sink_bucket='your-sink-bucket',
67    #     schedule_start_date=datetime.date.today()
68    # )
69    pass