mirror of
				https://github.com/hwchase17/langchain.git
				synced 2025-10-31 07:41:40 +00:00 
			
		
		
		
	* standardizes ruff dep version across all `pyproject.toml` files * cli: ruff rules and corrections * langchain: rules and corrections
		
			
				
	
	
		
			42 lines
		
	
	
		
			963 B
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			42 lines
		
	
	
		
			963 B
		
	
	
	
		
			Python
		
	
	
	
	
	
| import http.client
 | |
| import json
 | |
| from typing import Any, Optional, TypedDict
 | |
| 
 | |
| WRITE_KEY = "310apTK0HUFl4AOv"
 | |
| 
 | |
| 
 | |
| class EventDict(TypedDict):
 | |
|     event: str
 | |
|     properties: Optional[dict[str, Any]]
 | |
| 
 | |
| 
 | |
| def create_events(events: list[EventDict]) -> Optional[Any]:
 | |
|     try:
 | |
|         data = {
 | |
|             "events": [
 | |
|                 {
 | |
|                     "write_key": WRITE_KEY,
 | |
|                     "name": event["event"],
 | |
|                     "properties": event.get("properties"),
 | |
|                 }
 | |
|                 for event in events
 | |
|             ],
 | |
|         }
 | |
| 
 | |
|         conn = http.client.HTTPSConnection("app.firstpartyhq.com")
 | |
| 
 | |
|         payload = json.dumps(data)
 | |
| 
 | |
|         headers = {
 | |
|             "Content-Type": "application/json",
 | |
|             "Accept": "application/json",
 | |
|         }
 | |
| 
 | |
|         conn.request("POST", "/events/v1/track/bulk", payload, headers)
 | |
| 
 | |
|         res = conn.getresponse()
 | |
| 
 | |
|         return json.loads(res.read())
 | |
|     except Exception:
 | |
|         return None
 |