tessl/pypi-google-cloud-talent

Google Cloud Talent API client library for job search and talent management

—

Pending

Overview

Eval results

Files

Event Tracking and Analytics

Name: tessl/pypi-google-cloud-talent
Author: tessl

User interaction event tracking for improving search quality and providing analytics insights, including job views, applications, and other user behaviors with support for custom event metadata. The event system enables machine learning improvements and provides valuable insights into user engagement patterns.

Capabilities

Client Event Creation

Records user interaction events that help improve search quality and provide analytics insights for understanding user behavior and job performance.

def create_client_event(self, parent: str, client_event: ClientEvent) -> ClientEvent:
    """
    Records a client interaction event for analytics and search improvement.

    Parameters:
    - parent (str): Tenant resource name where event will be recorded
    - client_event (ClientEvent): Event object with interaction details

    Returns:
    ClientEvent: Recorded event with server-generated metadata

    Raises:
    - InvalidArgument: Missing required fields or invalid event data
    - PermissionDenied: Insufficient permissions to create events
    """

Usage Example:

from google.cloud.talent import EventServiceClient, ClientEvent, JobEvent
from google.protobuf.timestamp_pb2 import Timestamp
from datetime import datetime
import uuid

client = EventServiceClient()

# Record a job view event
job_view_event = ClientEvent(
    request_id="search-session-123",
    event_id=str(uuid.uuid4()),
    create_time=Timestamp(seconds=int(datetime.utcnow().timestamp())),
    job_event=JobEvent(
        type_=JobEvent.JobEventType.VIEW,
        jobs=["projects/my-project/tenants/my-tenant/jobs/job-123"]
    ),
    event_notes="User viewed job from search results page 1"
)

recorded_event = client.create_client_event(
    parent="projects/my-project/tenants/my-tenant",
    client_event=job_view_event
)

Event Data Model

Client Event

class ClientEvent:
    """
    User interaction event for analytics and search improvement.
    """
    request_id: str = None  # Unique request identifier linking to search requests
    event_id: str = None  # Unique event identifier (client-generated)
    create_time: Timestamp = None  # Event timestamp (client-generated)
    job_event: JobEvent = None  # Job-specific event details
    event_notes: str = None  # Additional event information (max 1000 chars)

Job Event

class JobEvent:
    """
    Job-specific user interaction events with detailed behavioral tracking.
    """
    type_: JobEventType = None  # Type of job interaction
    jobs: List[str] = None  # Job resource names involved in event (max 500)

class JobEventType(Enum):
    """Types of job-related user interactions."""
    JOB_EVENT_TYPE_UNSPECIFIED = 0
    IMPRESSION = 1  # Job appeared in search results
    VIEW = 2  # User viewed job details
    VIEW_REDIRECT = 3  # User clicked through to external job posting
    APPLICATION_START = 4  # User began job application process
    APPLICATION_FINISH = 5  # User completed job application
    APPLICATION_QUICK_SUBMISSION = 6  # User used quick/one-click application
    APPLICATION_REDIRECT = 7  # User redirected to external application system
    APPLICATION_START_FROM_SEARCH = 8  # Application started directly from search results
    APPLICATION_REDIRECT_FROM_SEARCH = 9  # Application redirect directly from search results
    APPLICATION_COMPANY_SUBMIT = 10  # Application submitted to company system
    BOOKMARK = 11  # User bookmarked/saved job
    NOTIFICATION = 12  # Push notification sent about job
    HIRED = 13  # User was hired for this job
    SENT_CV = 14  # User sent CV/resume for job
    INTERVIEW_GRANTED = 15  # User granted interview for job

Event Tracking Patterns

Search Session Tracking

Track complete user search sessions with multiple events linked by request_id.

import uuid
from datetime import datetime

# Generate session ID for tracking related events
session_id = f"search-session-{uuid.uuid4()}"

# 1. Record search impression events when jobs appear in results
for job_id in search_result_job_ids:
    impression_event = ClientEvent(
        request_id=session_id,
        event_id=str(uuid.uuid4()),
        create_time=Timestamp(seconds=int(datetime.utcnow().timestamp())),
        job_event=JobEvent(
            type_=JobEvent.JobEventType.IMPRESSION,
            jobs=[job_id]
        ),
        event_notes=f"Job appeared in search results position {position}"
    )
    client.create_client_event(parent=tenant_name, client_event=impression_event)

# 2. Record view events when user clicks on job details
view_event = ClientEvent(
    request_id=session_id,
    event_id=str(uuid.uuid4()),
    create_time=Timestamp(seconds=int(datetime.utcnow().timestamp())),
    job_event=JobEvent(
        type_=JobEvent.JobEventType.VIEW,
        jobs=["projects/my-project/tenants/my-tenant/jobs/job-456"]
    ),
    event_notes="User clicked job title from search results"
)
client.create_client_event(parent=tenant_name, client_event=view_event)

# 3. Record application events when user applies
application_event = ClientEvent(
    request_id=session_id,
    event_id=str(uuid.uuid4()), 
    create_time=Timestamp(seconds=int(datetime.utcnow().timestamp())),
    job_event=JobEvent(
        type_=JobEvent.JobEventType.APPLICATION_START,
        jobs=["projects/my-project/tenants/my-tenant/jobs/job-456"]
    ),
    event_notes="User clicked apply button"
)
client.create_client_event(parent=tenant_name, client_event=application_event)

Application Funnel Tracking

Track the complete application process from initial interest to hiring.

def track_application_funnel(job_id: str, user_id: str):
    """Track complete application funnel for analytics."""
    base_request_id = f"application-{user_id}-{job_id}"
    
    # Application started
    start_event = ClientEvent(
        request_id=base_request_id,
        event_id=str(uuid.uuid4()),
        create_time=Timestamp(seconds=int(datetime.utcnow().timestamp())),
        job_event=JobEvent(
            type_=JobEvent.JobEventType.APPLICATION_START,
            jobs=[job_id]
        ),
        event_notes="User initiated application process"
    )
    client.create_client_event(parent=tenant_name, client_event=start_event)
    
    # Application completed (call when form submitted)
    def track_application_completion():
        finish_event = ClientEvent(
            request_id=base_request_id,
            event_id=str(uuid.uuid4()),
            create_time=Timestamp(seconds=int(datetime.utcnow().timestamp())),
            job_event=JobEvent(
                type_=JobEvent.JobEventType.APPLICATION_FINISH,
                jobs=[job_id]
            ),
            event_notes="User completed application form"
        )
        client.create_client_event(parent=tenant_name, client_event=finish_event)
    
    # CV/Resume sent (call when resume uploaded/sent)
    def track_cv_sent():
        cv_event = ClientEvent(
            request_id=base_request_id,
            event_id=str(uuid.uuid4()),
            create_time=Timestamp(seconds=int(datetime.utcnow.timestamp())),
            job_event=JobEvent(
                type_=JobEvent.JobEventType.SENT_CV,
                jobs=[job_id]
            ),
            event_notes="User resume sent to employer"
        )
        client.create_client_event(parent=tenant_name, client_event=cv_event)
    
    return track_application_completion, track_cv_sent

Batch Event Recording

For high-volume applications, consider batching events to reduce API calls.

class EventBatcher:
    def __init__(self, tenant_name: str, batch_size: int = 100):
        self.tenant_name = tenant_name
        self.batch_size = batch_size
        self.pending_events = []
        self.client = EventServiceClient()
    
    def add_event(self, event: ClientEvent):
        """Add event to batch queue."""
        self.pending_events.append(event)
        
        if len(self.pending_events) >= self.batch_size:
            self.flush()
    
    def flush(self):
        """Send all pending events."""
        for event in self.pending_events:
            try:
                self.client.create_client_event(
                    parent=self.tenant_name,
                    client_event=event
                )
            except Exception as e:
                print(f"Failed to send event {event.event_id}: {e}")
        
        self.pending_events.clear()

# Usage
batcher = EventBatcher("projects/my-project/tenants/my-tenant")

# Add events throughout user session
batcher.add_event(impression_event)
batcher.add_event(view_event)
batcher.add_event(application_event)

# Ensure all events are sent at session end
batcher.flush()

Event Request and Response Types

Event Service Requests

class CreateClientEventRequest:
    parent: str = None  # Tenant resource name
    client_event: ClientEvent = None  # Event to record

Event Analytics Integration

Events are processed by Google's machine learning systems to improve search quality and provide insights. While the API doesn't expose analytics directly, recorded events contribute to:

Search Ranking: Jobs with higher engagement get better ranking
Query Understanding: User interactions help improve query interpretation
Personalization: Individual user preferences for future searches
Company Analytics: Aggregate performance metrics for employers

Event Metadata and Context

Request Metadata Integration

Link events to search requests using RequestMetadata for enhanced analytics.

from google.cloud.talent import RequestMetadata, DeviceInfo

# Create consistent request metadata for search and events
request_metadata = RequestMetadata(
    domain="mycompany.com",
    session_id="user-session-789",
    user_id="user-12345",
    device_info=DeviceInfo(
        device_type=DeviceInfo.DeviceType.WEB,
        id="device-id-browser-123"
    )
)

# Use same session_id in events
event = ClientEvent(
    request_id="user-session-789",  # Match search session
    event_id=str(uuid.uuid4()),
    # ... other event fields
)

Custom Event Notes

Use event_notes field to provide additional context for analytics and debugging.

# Include contextual information in event notes
event_notes_examples = [
    "Job clicked from search results position 3",
    "Application started from job detail page",
    "Quick apply used - LinkedIn integration", 
    "User bookmarked job for later review",
    "Push notification opened leading to job view",
    "Email campaign click-through to job posting"
]

event = ClientEvent(
    # ... other fields
    event_notes="User applied after viewing job description for 2+ minutes"
)

Error Handling

Event tracking operations can raise several types of exceptions:

from google.api_core import exceptions
import logging

def safe_track_event(client: EventServiceClient, parent: str, event: ClientEvent):
    """Safely track event with error handling."""
    try:
        return client.create_client_event(parent=parent, client_event=event)
    except exceptions.InvalidArgument as e:
        logging.error(f"Invalid event data: {e}")
        # Handle validation errors - check required fields
    except exceptions.PermissionDenied as e:
        logging.error(f"Event tracking permission denied: {e}")
        # Handle authorization errors
    except exceptions.ResourceExhausted as e:
        logging.warning(f"Event tracking quota exceeded: {e}")
        # Handle quota limits - consider backing off
    except exceptions.DeadlineExceeded as e:
        logging.warning(f"Event tracking timeout: {e}")
        # Handle timeouts - retry or skip
    except Exception as e:
        logging.error(f"Unexpected event tracking error: {e}")
        # Handle unexpected errors gracefully
        
    return None  # Event tracking failed but don't break user flow

Common error scenarios:

InvalidArgument: Missing required fields (event_id, job_event), invalid timestamps
PermissionDenied: Insufficient IAM permissions for event creation
ResourceExhausted: Event tracking API quota limits exceeded
DeadlineExceeded: Event submission timeout

Best Practices

Event Timing: Record events as close to when they occur as possible for accuracy
Unique IDs: Always use unique event_id values to prevent duplicate event processing
Session Linking: Use consistent request_id values to link related events in a user session
Error Handling: Never let event tracking failures break the user experience
Privacy: Follow privacy laws and company policies when tracking user interactions
Event Volume: Consider batching for high-volume applications to manage API quotas
Testing: Implement event tracking testing to ensure data quality
Monitoring: Monitor event tracking success rates and API quota usage

Common Integration Patterns

React/JavaScript Frontend

// Frontend event tracking integration
class TalentEventTracker {
    constructor(sessionId) {
        this.sessionId = sessionId;
        this.eventQueue = [];
    }
    
    trackJobImpression(jobId, position) {
        this.queueEvent({
            type: 'IMPRESSION',
            jobId: jobId,
            metadata: { position: position }
        });
    }
    
    trackJobView(jobId) {
        this.queueEvent({
            type: 'VIEW', 
            jobId: jobId,
            metadata: { timestamp: Date.now() }
        });
        
        // Send immediately for important events
        this.flush();
    }
    
    trackApplicationStart(jobId) {
        this.queueEvent({
            type: 'APPLICATION_START',
            jobId: jobId,
            metadata: { timestamp: Date.now() }
        });
        this.flush();
    }
    
    queueEvent(event) {
        this.eventQueue.push({
            ...event,
            sessionId: this.sessionId,
            eventId: this.generateEventId()
        });
    }
    
    async flush() {
        if (this.eventQueue.length === 0) return;
        
        try {
            await fetch('/api/talent/events', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({
                    events: this.eventQueue.splice(0)
                })
            });
        } catch (error) {
            console.warn('Event tracking failed:', error);
        }
    }
    
    generateEventId() {
        return 'event_' + Date.now() + '_' + Math.random().toString(36).substr(2, 9);
    }
}

Analytics Dashboard Integration

class TalentAnalytics:
    """Helper class for integrating with analytics dashboards."""
    
    def __init__(self, tenant_name: str):
        self.tenant_name = tenant_name
        self.client = EventServiceClient()
    
    def track_search_to_application_funnel(self, search_session_id: str, job_ids: List[str]):
        """Track complete search-to-application funnel."""
        funnel_stages = [
            (JobEvent.JobEventType.IMPRESSION, "Search results displayed"),
            (JobEvent.JobEventType.VIEW, "Job details viewed"),
            (JobEvent.JobEventType.APPLICATION_START, "Application process started"),
            (JobEvent.JobEventType.APPLICATION_FINISH, "Application completed")
        ]
        
        for stage_type, description in funnel_stages:
            event = ClientEvent(
                request_id=search_session_id,
                event_id=str(uuid.uuid4()),
                create_time=Timestamp(seconds=int(datetime.utcnow().timestamp())),
                job_event=JobEvent(type_=stage_type, jobs=job_ids),
                event_notes=description
            )
            
            self.client.create_client_event(
                parent=self.tenant_name,
                client_event=event
            )

Install with Tessl CLI

npx tessl i tessl/pypi-google-cloud-talent

docs

company-management.md

event-tracking.md

index.md

job-management.md

query-autocompletion.md

search-filtering.md

tenant-management.md

tile.json