Job Management

def create_job(self, parent: str, job: Job) -> Job:
    """
    Creates a new job posting.

    Parameters:
    - parent (str): Tenant resource name where job will be created
    - job (Job): Job object with required fields (company, requisition_id, title, description)

    Returns:
    Job: Created job with generated resource name and derived information

    Raises:
    - InvalidArgument: Missing required fields or invalid values
    - AlreadyExists: Job with same requisition_id already exists
    - PermissionDenied: Insufficient permissions to create job
    """

from google.cloud.talent import JobServiceClient, Job, EmploymentType, JobLevel

client = JobServiceClient()

job = Job(
    company="projects/my-project/tenants/my-tenant/companies/company-123",
    requisition_id="req-software-eng-001",
    title="Senior Software Engineer",
    description="""
    <p>Join our engineering team to build scalable cloud solutions.</p>
    <h3>Requirements:</h3>
    <ul>
        <li>5+ years of Python experience</li>
        <li>Experience with cloud platforms</li>
        <li>Strong problem-solving skills</li>
    </ul>
    """,
    addresses=["1600 Amphitheatre Parkway, Mountain View, CA 94043"],
    employment_types=[EmploymentType.FULL_TIME],
    job_level=JobLevel.EXPERIENCED,
    qualifications="Bachelor's degree in Computer Science or equivalent experience"
)

created_job = client.create_job(
    parent="projects/my-project/tenants/my-tenant",
    job=job
)

def get_job(self, name: str) -> Job:
    """
    Retrieves a job by its resource name.

    Parameters:
    - name (str): Full job resource name

    Returns:
    Job: Complete job object with all fields

    Raises:
    - NotFound: Job does not exist
    - PermissionDenied: Insufficient permissions to view job
    """

def update_job(self, job: Job, update_mask: FieldMask = None) -> Job:
    """
    Updates an existing job posting.

    Parameters:
    - job (Job): Job object with updated values and resource name
    - update_mask (FieldMask): Specifies which fields to update

    Returns:
    Job: Updated job object

    Raises:
    - NotFound: Job does not exist
    - InvalidArgument: Invalid field values or update mask
    - PermissionDenied: Insufficient permissions to update job
    """

from google.protobuf import field_mask_pb2

# Update job title and description only
job.title = "Staff Software Engineer"
job.description = "Updated job description with new requirements"

update_mask = field_mask_pb2.FieldMask(paths=["title", "description"])

updated_job = client.update_job(job=job, update_mask=update_mask)

def delete_job(self, name: str) -> None:
    """
    Deletes a job posting.

    Parameters:
    - name (str): Full job resource name

    Raises:
    - NotFound: Job does not exist
    - PermissionDenied: Insufficient permissions to delete job
    """

def list_jobs(self, parent: str, filter: str = None, page_size: int = None, 
              page_token: str = None, job_view: JobView = None) -> ListJobsResponse:
    """
    Lists jobs with optional filtering and pagination.

    Parameters:
    - parent (str): Tenant resource name
    - filter (str): Filter expression for job attributes
    - page_size (int): Maximum number of jobs to return (max 1000)
    - page_token (str): Token for pagination from previous response
    - job_view (JobView): Level of detail to return

    Returns:
    ListJobsResponse: Jobs list with pagination token and metadata

    Raises:
    - InvalidArgument: Invalid filter expression or page size
    - PermissionDenied: Insufficient permissions to list jobs
    """

from google.cloud.talent import JobView

# List all jobs with minimal details
response = client.list_jobs(
    parent="projects/my-project/tenants/my-tenant",
    job_view=JobView.JOB_VIEW_MINIMAL,
    page_size=50
)

for job in response.jobs:
    print(f"Job: {job.title} at {job.company}")

# Continue pagination if needed
if response.next_page_token:
    next_response = client.list_jobs(
        parent="projects/my-project/tenants/my-tenant",
        page_token=response.next_page_token,
        page_size=50
    )

def search_jobs(self, request: SearchJobsRequest) -> SearchJobsResponse:
    """
    Searches for jobs with advanced filtering and ranking.

    Parameters:
    - request (SearchJobsRequest): Complete search request with query, filters, and options

    Returns:
    SearchJobsResponse: Matching jobs with ranking, metadata, and histogram results

    Raises:
    - InvalidArgument: Invalid search parameters or query syntax
    - PermissionDenied: Insufficient permissions to search jobs
    """

from google.cloud.talent import (
    SearchJobsRequest, JobQuery, LocationFilter, CompensationFilter,
    HistogramQuery, SearchMode
)

search_request = SearchJobsRequest(
    parent="projects/my-project/tenants/my-tenant",
    search_mode=SearchMode.JOB_SEARCH,
    job_query=JobQuery(
        query="python software engineer",
        location_filters=[
            LocationFilter(
                address="San Francisco, CA",
                distance_in_miles=25.0
            )
        ],
        compensation_filter=CompensationFilter(
            type_=CompensationFilter.FilterType.ANNUALIZED_BASE_AMOUNT,
            range_=CompensationRange(
                max_compensation=Money(currency_code="USD", units=150000),
                min_compensation=Money(currency_code="USD", units=100000)
            )
        )
    ),
    histogram_queries=[
        HistogramQuery(histogram_query="employment_type"),
        HistogramQuery(histogram_query="company_size")
    ],
    job_view=JobView.JOB_VIEW_SMALL,
    page_size=20
)

search_response = client.search_jobs(search_request)

for matching_job in search_response.matching_jobs:
    job = matching_job.job
    print(f"Job: {job.title} at {job.company}")
    print(f"Match score: {matching_job.match_type}")

# Process histogram results for faceted search UI
for histogram_result in search_response.histogram_query_results:
    print(f"Facet: {histogram_result.histogram_query}")
    for bucket in histogram_result.histogram:
        print(f"  {bucket.range_}: {bucket.count} jobs")

def search_jobs_for_alert(self, request: SearchJobsRequest) -> SearchJobsResponse:
    """
    Searches for jobs with algorithms optimized for passive job seekers.
    
    This method is specifically designed for targeting passive job seekers 
    (e.g., users signed up for job alert emails) with algorithmic adjustments
    that differ from active job search scenarios.

    Parameters:
    - request (SearchJobsRequest): Search request with query and filtering options

    Returns:
    SearchJobsResponse: Jobs ranked for passive job seeker scenarios

    Raises:
    - InvalidArgument: Invalid search parameters
    - PermissionDenied: Insufficient permissions to search jobs
    """

from google.cloud.talent import SearchJobsRequest, JobQuery, LocationFilter

# Create search request optimized for job alerts
alert_request = SearchJobsRequest(
    parent="projects/my-project/tenants/my-tenant",
    job_query=JobQuery(
        query="software engineer python",
        location_filters=[
            LocationFilter(
                address="San Francisco Bay Area, CA",
                distance_in_miles=50.0
            )
        ]
    ),
    job_view=JobView.JOB_VIEW_SMALL,
    page_size=10
)

# Search with alert-optimized algorithms
alert_response = client.search_jobs_for_alert(alert_request)

for matching_job in alert_response.matching_jobs:
    job = matching_job.job
    print(f"Alert job: {job.title} at {job.company}")

def batch_create_jobs(self, parent: str, jobs: List[Job]) -> Operation:
    """
    Creates multiple jobs in a single batch operation.

    Parameters:
    - parent (str): Tenant resource name
    - jobs (List[Job]): List of jobs to create (max 200)

    Returns:
    Operation: Long-running operation for tracking batch progress

    Raises:
    - InvalidArgument: Too many jobs or invalid job data
    - PermissionDenied: Insufficient permissions for batch operations
    """

def batch_update_jobs(self, parent: str, jobs: List[Job]) -> Operation:
    """
    Updates multiple jobs in a single batch operation.

    Parameters:
    - parent (str): Tenant resource name
    - jobs (List[Job]): List of jobs to update with resource names (max 200)

    Returns:
    Operation: Long-running operation for tracking batch progress
    """

def batch_delete_jobs(self, parent: str, names: List[str]) -> Operation:
    """
    Deletes multiple jobs in a single batch operation.

    Parameters:
    - parent (str): Tenant resource name
    - names (List[str]): List of job resource names to delete (max 200)

    Returns:
    Operation: Long-running operation for tracking batch progress
    """

from google.cloud.talent import Job
import time

# Batch create multiple jobs
jobs_to_create = []
for i in range(10):
    job = Job(
        company="projects/my-project/tenants/my-tenant/companies/company-123",
        requisition_id=f"batch-job-{i}",
        title=f"Software Engineer {i}",
        description="Batch created job posting",
        addresses=["Remote"]
    )
    jobs_to_create.append(job)

# Start batch creation
operation = client.batch_create_jobs(
    parent="projects/my-project/tenants/my-tenant",
    jobs=jobs_to_create
)

# Monitor operation progress
while not operation.done():
    print("Batch operation in progress...")
    time.sleep(5)
    operation = client.get_operation(name=operation.name)

# Get final results
if operation.done():
    if operation.error:
        print(f"Batch operation failed: {operation.error}")
    else:
        result = operation.result()
        print(f"Successfully created {len(result.job_results)} jobs")

def get_operation(self, request: GetOperationRequest) -> Operation:
    """
    Gets the status of a long-running operation.

    Parameters:
    - request (GetOperationRequest): Operation name and metadata

    Returns:
    Operation: Current operation status with progress and results

    Raises:
    - NotFound: Operation does not exist
    - PermissionDenied: Insufficient permissions to view operation
    """

class CreateJobRequest:
    parent: str = None  # Tenant resource name
    job: Job = None  # Job to create

class GetJobRequest:
    name: str = None  # Job resource name

class UpdateJobRequest:
    job: Job = None  # Job with updates
    update_mask: FieldMask = None  # Fields to update

class DeleteJobRequest:
    name: str = None  # Job resource name

class ListJobsRequest:
    parent: str = None  # Tenant resource name
    filter: str = None  # Filter expression
    page_size: int = None  # Page size (max 1000)
    page_token: str = None  # Pagination token
    job_view: JobView = None  # Detail level

class SearchJobsRequest:
    parent: str = None  # Tenant resource name
    search_mode: SearchMode = None  # Search mode
    job_query: JobQuery = None  # Search query and filters
    enable_broadening: bool = None  # Enable query broadening
    histogram_queries: List[HistogramQuery] = None  # Faceted search
    job_view: JobView = None  # Detail level
    offset: int = None  # Result offset
    page_size: int = None  # Page size (max 100)
    page_token: str = None  # Pagination token
    order_by: str = None  # Sort order
    diversification_level: DiversificationLevel = None  # Result diversity
    custom_ranking_info: CustomRankingInfo = None  # Custom ranking

class ListJobsResponse:
    jobs: List[Job] = None  # List of jobs
    next_page_token: str = None  # Pagination token
    metadata: ResponseMetadata = None  # Response metadata

class SearchJobsResponse:
    matching_jobs: List[MatchingJob] = None  # Jobs with match info
    histogram_query_results: List[HistogramQueryResult] = None  # Facet results
    next_page_token: str = None  # Pagination token
    location_filters: List[Location] = None  # Matched locations
    spell_correction: SpellCheckResult = None  # Query corrections
    metadata: ResponseMetadata = None  # Response metadata
    total_size: int = None  # Total result count
    broadened_query_jobs_count: int = None  # Broadened results count

class BatchCreateJobsResponse:
    job_results: List[JobResult] = None  # Individual job results

class BatchUpdateJobsResponse:
    job_results: List[JobResult] = None  # Individual job results

class BatchDeleteJobsResponse:
    job_results: List[JobResult] = None  # Individual job results

class JobResult:
    job: Job = None  # Job data
    status: Status = None  # Operation status

class MatchingJob:
    job: Job = None  # Matched job
    job_summary: str = None  # Highlighted summary
    job_title_snippet: str = None  # Highlighted title
    search_text_snippet: str = None  # Highlighted text
    commute_info: CommuteInfo = None  # Commute information

class JobView(Enum):
    JOB_VIEW_UNSPECIFIED = 0
    JOB_VIEW_ID_ONLY = 1
    JOB_VIEW_MINIMAL = 2
    JOB_VIEW_SMALL = 3
    JOB_VIEW_FULL = 4

class SearchMode(Enum):
    JOB_SEARCH = 1
    FEATURED_JOB_SEARCH = 2

from google.api_core import exceptions

try:
    job = client.create_job(parent=parent, job=job_data)
except exceptions.InvalidArgument as e:
    # Handle validation errors
    print(f"Invalid job data: {e}")
except exceptions.AlreadyExists as e:
    # Handle duplicate requisition_id
    print(f"Job already exists: {e}")
except exceptions.ResourceExhausted as e:
    # Handle quota limits
    print(f"Quota exceeded: {e}")
except exceptions.PermissionDenied as e:
    # Handle authorization errors
    print(f"Access denied: {e}")