blob: a083ede21cc7a1023c656507bd2422f6a738fa93 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
import os
import time
from pathlib import Path
from dotenv import load_dotenv
import httpx
import supermemory
from supermemory import Supermemory
load_dotenv()
client = Supermemory()
def test_documents_crud():
"""Test document CRUD operations"""
print("\n=== Document CRUD Operations ===")
# Create
doc = client.documents.add(content=f"Test content - {time.time()}")
print(f"✓ documents.add: {doc.id}")
# Read
fetched = client.documents.get(doc.id)
print(f"✓ documents.get: {fetched.id}")
# Update
updated = client.documents.update(doc.id, content=f"Updated - {time.time()}")
print(f"✓ documents.update: {updated.id}")
# Wait for processing
time.sleep(10)
# Delete
client.documents.delete(doc.id)
print("✓ documents.delete")
def test_batch_operations():
"""Test batch operations"""
print("\n=== Batch Operations ===")
batch = client.documents.batch_add(
documents=[
{"content": f"Batch 1 - {time.time()}"},
{"content": f"Batch 2 - {time.time()}"},
]
)
print(f"✓ documents.batch_add: {batch}")
def test_search():
"""Test search"""
print("\n=== Search ===")
results = client.search.execute(q="test content")
print(f"✓ search.execute: {len(results.results) if results.results else 0} results")
def test_file_uploads():
"""Test file uploads"""
print("\n=== File Uploads ===")
test_path = Path("/tmp/test-py-upload.txt")
test_path.write_text(f"Test content {time.time()}")
client.documents.upload_file(file=test_path)
print("✓ documents.upload_file with Path")
test_path.unlink()
def test_error_handling():
"""Test error handling patterns"""
print("\n=== Error Handling ===")
try:
client.documents.add(content="Test content")
print("✓ Error handling pattern works")
except supermemory.APIConnectionError:
print("APIConnectionError handled")
except supermemory.RateLimitError:
print("RateLimitError handled")
except supermemory.APIStatusError:
print("APIStatusError handled")
def test_client_config():
"""Test client configuration"""
print("\n=== Client Configuration ===")
# Retries
client2 = Supermemory(max_retries=0)
print("✓ max_retries config")
# Timeout
client3 = Supermemory(timeout=20.0)
print("✓ timeout config")
# Granular timeout
client4 = Supermemory(timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0))
print("✓ granular timeout config")
# Per-request options
client.with_options(max_retries=5).documents.add(content="Test content")
print("✓ per-request options")
def test_raw_response():
"""Test raw response access"""
print("\n=== Raw Response ===")
response = client.documents.with_raw_response.add(content="Test content")
print(f"✓ with_raw_response: has headers={hasattr(response, 'headers')}")
memory = response.parse()
print(f"✓ parse(): {memory.id}")
def test_streaming_response():
"""Test streaming response"""
print("\n=== Streaming Response ===")
with client.documents.with_streaming_response.add(content="Test content") as response:
print(f"✓ with_streaming_response: has headers={hasattr(response, 'headers')}")
def test_context_manager():
"""Test context manager"""
print("\n=== Context Manager ===")
with Supermemory() as temp_client:
print("✓ Context manager works")
def main():
print("Python SDK Tests")
print("================")
test_documents_crud()
test_batch_operations()
test_search()
test_file_uploads()
test_error_handling()
test_client_config()
test_raw_response()
test_streaming_response()
test_context_manager()
print("\n================")
print("✅ All Python SDK tests passed!")
if __name__ == "__main__":
main()
|