0
# Blob Management
1

2
The Blob class represents individual objects stored in Google Cloud Storage, providing methods for content manipulation, metadata management, and secure access through signed URLs. Blob instances are created through Bucket operations and maintain references to their parent bucket and storage client.
3

4
## Capabilities
5

6
### Blob Initialization
7

8
Blob instances are typically created through Bucket methods like `get_blob()` or `new_blob()` rather than direct instantiation.
9

10
```python { .api }
11
def __init__(self, bucket, name, metadata):
12
    """
13
    Initialize a Blob instance.
14
    
15
    Parameters:
16
    - bucket (Bucket): Parent bucket instance
17
    - name (str): Blob name/path
18
    - metadata (Dict[str, Any]): Blob metadata from GCS
19
    
20
    Attributes:
21
    - bucket (Bucket): Reference to parent bucket
22
    - name (str): Blob name/path
23
    - size (int): Blob size in bytes (from metadata)
24
    - chunk_size (int): Download chunk size for streaming
25
    """
26
```
27

28
**Usage Example:**
29
```python
30
async with Storage() as storage:
31
    bucket = storage.get_bucket('my-bucket')
32
    
33
    # Get existing blob with metadata
34
    blob = await bucket.get_blob('existing-file.txt')
35
    
36
    # Create new blob instance (without metadata)
37
    new_blob = bucket.new_blob('new-file.txt')
38
    
39
    # Access blob properties
40
    print(f"Blob name: {blob.name}")
41
    print(f"Blob size: {blob.size} bytes")
42
    print(f"Parent bucket: {blob.bucket.name}")
43
```
44

45
### Content Download Operations
46

47
Download blob content with options for automatic decompression and different return formats.
48

49
```python { .api }
50
async def download(self, timeout=10, session=None, auto_decompress=True):
51
    """
52
    Download blob content.
53
    
54
    Parameters:
55
    - timeout (int): Request timeout in seconds
56
    - session (aiohttp.ClientSession, optional): Custom session
57
    - auto_decompress (bool): Automatically decompress gzip content
58
    
59
    Returns:
60
    Any: Blob content (typically bytes, but can be decompressed text)
61
    """
62
```
63

64
**Usage Example:**
65
```python
66
async with Storage() as storage:
67
    bucket = storage.get_bucket('my-bucket')
68
    blob = await bucket.get_blob('data.json')
69
    
70
    # Download content as bytes
71
    content = await blob.download()
72
    
73
    # Download with custom timeout
74
    content = await blob.download(timeout=30)
75
    
76
    # Download without auto-decompression for gzipped content
77
    raw_content = await blob.download(auto_decompress=False)
78
    
79
    # Convert to string if needed
80
    if isinstance(content, bytes):
81
        text_content = content.decode('utf-8')
82
```
83

84
### Content Upload Operations
85

86
Upload data to the blob with content type specification and metadata handling.
87

88
```python { .api }
89
async def upload(self, data, content_type=None, session=None):
90
    """
91
    Upload data to the blob.
92
    
93
    Parameters:
94
    - data (bytes or file-like): Data to upload
95
    - content_type (str, optional): MIME type of the content
96
    - session (aiohttp.ClientSession, optional): Custom session
97
    
98
    Returns:
99
    Dict[str, Any]: Upload response metadata
100
    """
101
```
102

103
**Usage Example:**
104
```python
105
async with Storage() as storage:
106
    bucket = storage.get_bucket('my-bucket')
107
    
108
    # Create new blob and upload data
109
    blob = bucket.new_blob('new-document.pdf')
110
    with open('/local/path/document.pdf', 'rb') as f:
111
        pdf_data = f.read()
112
    
113
    result = await blob.upload(pdf_data, content_type='application/pdf')
114
    
115
    # Upload text content
116
    text_blob = bucket.new_blob('notes.txt')
117
    text_data = "Important notes here".encode('utf-8')
118
    result = await text_blob.upload(text_data, content_type='text/plain')
119
    
120
    # Upload JSON data
121
    json_blob = bucket.new_blob('config.json')
122
    import json
123
    json_data = json.dumps({'setting': 'value'}).encode('utf-8')
124
    result = await json_blob.upload(json_data, content_type='application/json')
125
```
126

127
### Signed URL Generation
128

129
Generate temporary, secure URLs for accessing blob content without authentication credentials.
130

131
```python { .api }
132
async def get_signed_url(self, expiration, headers=None, query_params=None, http_method='GET', iam_client=None, service_account_email=None, token=None, session=None):
133
    """
134
    Generate a signed URL for the blob.
135
    
136
    Parameters:
137
    - expiration (int or datetime): URL expiration (seconds from now or datetime object)
138
    - headers (dict, optional): Headers to include in the signed request
139
    - query_params (dict, optional): Query parameters to include
140
    - http_method (str): HTTP method the URL will be used with
141
    - iam_client (IAMClient, optional): IAM client for signature generation
142
    - service_account_email (str, optional): Service account for IAM signing
143
    - token (Token, optional): Authentication token
144
    - session (aiohttp.ClientSession, optional): Custom session
145
    
146
    Returns:
147
    str: Signed URL for blob access
148
    """
149
```
150

151
**Usage Example:**
152
```python
153
from datetime import datetime, timedelta
154

155
async with Storage() as storage:
156
    bucket = storage.get_bucket('my-bucket')
157
    blob = await bucket.get_blob('private-document.pdf')
158
    
159
    # Generate URL valid for 1 hour (3600 seconds)
160
    url = await blob.get_signed_url(3600)
161
    print(f"Temporary URL: {url}")
162
    
163
    # Generate URL with specific expiration datetime
164
    expiry = datetime.now() + timedelta(hours=2)
165
    url = await blob.get_signed_url(expiry)
166
    
167
    # Generate URL for POST uploads with custom headers
168
    headers = {'Content-Type': 'application/pdf'}
169
    query_params = {'uploadType': 'media'}
170
    upload_url = await blob.get_signed_url(
171
        3600, 
172
        headers=headers,
173
        query_params=query_params,
174
        http_method='POST'
175
    )
176
    
177
    # Generate URL using IAM service account
178
    url = await blob.get_signed_url(
179
        3600,
180
        service_account_email='my-service@project.iam.gserviceaccount.com'
181
    )
182
```
183

184
### Cryptographic Signature Methods
185

186
Static methods for generating cryptographic signatures used in signed URL creation.
187

188
```python { .api }
189
@staticmethod
190
def get_pem_signature(str_to_sign, private_key):
191
    """
192
    Generate PEM-based signature for string signing.
193
    
194
    Parameters:
195
    - str_to_sign (str): String to be signed
196
    - private_key (str): PEM-formatted private key
197
    
198
    Returns:
199
    bytes: Cryptographic signature
200
    """
201

202
@staticmethod
203
async def get_iam_api_signature(str_to_sign, iam_client, service_account_email, session):
204
    """
205
    Generate signature using Google IAM API.
206
    
207
    Parameters:
208
    - str_to_sign (str): String to be signed
209
    - iam_client (IAMClient): IAM client instance
210
    - service_account_email (str): Service account email
211
    - session (aiohttp.ClientSession): HTTP session
212
    
213
    Returns:
214
    bytes: Cryptographic signature from IAM API
215
    """
216
```
217

218
**Usage Example:**
219
```python
220
# These are typically used internally by get_signed_url()
221
# but can be used directly for custom signing scenarios
222

223
# Using PEM key directly
224
private_key_pem = """-----BEGIN PRIVATE KEY-----
225
MIIEvQ...
226
-----END PRIVATE KEY-----"""
227

228
signature = Blob.get_pem_signature("string to sign", private_key_pem)
229

230
# Using IAM API for signing (requires IAM client setup)
231
from gcloud.aio.auth import IAMClient
232

233
async with IAMClient() as iam_client:
234
    signature = await Blob.get_iam_api_signature(
235
        "string to sign",
236
        iam_client,
237
        "service-account@project.iam.gserviceaccount.com",
238
        session
239
    )
240
```
241

242
## Common Usage Patterns
243

244
### File Processing Pipeline
245

246
```python
247
async with Storage() as storage:
248
    bucket = storage.get_bucket('processing-bucket')
249
    
250
    # Get input blob
251
    input_blob = await bucket.get_blob('input/data.csv')
252
    csv_content = await input_blob.download()
253
    
254
    # Process data
255
    processed_data = process_csv(csv_content)
256
    
257
    # Upload processed result
258
    output_blob = bucket.new_blob('output/processed-data.json')
259
    json_data = json.dumps(processed_data).encode('utf-8')
260
    await output_blob.upload(json_data, content_type='application/json')
261
```
262

263
### Secure File Sharing
264

265
```python
266
async with Storage() as storage:
267
    bucket = storage.get_bucket('shared-files')
268
    blob = await bucket.get_blob('confidential-report.pdf')
269
    
270
    # Generate short-lived URL for sharing
271
    secure_url = await blob.get_signed_url(
272
        3600,  # 1 hour expiration
273
        http_method='GET'
274
    )
275
    
276
    # Send URL to authorized user
277
    send_secure_link(user_email, secure_url)
278
```
279

280
### Backup and Archival
281

282
```python
283
async with Storage() as storage:
284
    source_bucket = storage.get_bucket('live-data')
285
    archive_bucket = storage.get_bucket('archive')
286
    
287
    # Get blob from live bucket
288
    live_blob = await source_bucket.get_blob('important-data.db')
289
    content = await live_blob.download()
290
    
291
    # Create archived copy with timestamp
292
    from datetime import datetime
293
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
294
    archive_blob = archive_bucket.new_blob(f'backups/data_{timestamp}.db')
295
    await archive_blob.upload(content, content_type='application/octet-stream')
296
```