Utils Module

The utils module provides various utility functions and integrations for The Data Packet.

MongoDB Integration

class the_data_packet.utils.mongodb.MongoDBClient(username: str, password: str)

Bases: object

A MongoDB client wrapper for database operations.

This class provides a simplified interface for connecting to MongoDB and performing common database operations like inserting and querying documents.

client

The MongoDB client instance.

Type:

MongoClient

db

The MongoDB database instance.

Type:

Database

__init__(username: str, password: str) → None

Initialize the MongoDB client.

Parameters:

• username (str) – The username for MongoDB authentication.

• password (str) – The password for MongoDB authentication.

get_collection(collection_name: str) → Collection

Get a collection from the database.

Parameters:

collection_name (str) – The name of the collection to retrieve.

Returns:

The MongoDB collection instance.

Return type:

Collection

insert_document(collection_name: str, document: Dict[str, Any]) → InsertOneResult

Insert a single document into a collection.

Parameters:

• collection_name (str) – The name of the collection to insert into.

• document (Dict[str, Any]) – The document to insert.

Returns:

The result of the insert operation, containing

information about the insertion including the inserted_id.

Return type:

InsertOneResult

find_documents(collection_name: str, query: Dict[str, Any] | None = None) → Cursor

Find documents in a collection based on a query.

Parameters:

• collection_name (str) – The name of the collection to search in.

• query (Optional[Dict[str, Any]]) – The query filter. If None or empty, returns all documents in the collection.

Returns:

A cursor object that can be iterated over to access

the matching documents.

Return type:

Cursor

close() → None

Close the MongoDB client connection.

This should be called when the client is no longer needed to properly close the connection and free resources.

S3 Storage

AWS S3 storage backend.

class the_data_packet.utils.s3.S3UploadResult(success: bool, s3_url: str | None = None, error_message: str | None = None, file_size_bytes: int | None = None)

Bases: object

Result of S3 upload operation.

success: bool

s3_url: str | None = None

error_message: str | None = None

file_size_bytes: int | None = None

__init__(success: bool, s3_url: str | None = None, error_message: str | None = None, file_size_bytes: int | None = None) → None

class the_data_packet.utils.s3.S3Storage(bucket_name: str | None = None, aws_access_key_id: str | None = None, aws_secret_access_key: str | None = None, region: str | None = None)

Bases: object

AWS S3 storage backend.

__init__(bucket_name: str | None = None, aws_access_key_id: str | None = None, aws_secret_access_key: str | None = None, region: str | None = None)

Initialize S3 storage.

Parameters:

• bucket_name – S3 bucket name (defaults to config)

• aws_access_key_id – AWS access key (defaults to config/env)

• aws_secret_access_key – AWS secret key (defaults to config/env)

• region – AWS region (defaults to config)

upload_file(local_path: Path, s3_key: str | None = None, content_type: str | None = None) → S3UploadResult

Upload a file to S3.

Parameters:

• local_path – Path to local file

• s3_key – S3 object key (defaults to filename)

• public – Whether to make the file publicly readable

• content_type – MIME type for the file

Returns:

S3UploadResult with upload details

bucket_exists() → bool

Check if the configured bucket exists and is accessible.

HTTP Utils

HTTP client utility.

class the_data_packet.utils.http.HTTPClient(timeout: int | None = None, user_agent: str | None = None)

Bases: object

Simple HTTP client with error handling and configuration.

__init__(timeout: int | None = None, user_agent: str | None = None)

Initialize HTTP client.

Parameters:

• timeout – Request timeout (defaults to config)

• user_agent – User agent string (defaults to config)

get(url: str, **kwargs: Any) → Response

Make a GET request.

Parameters:

• url – URL to fetch

• **kwargs – Additional arguments passed to requests.get

Returns:

Response object

Raises:

NetworkError – If request fails

get_soup(url: str, **kwargs: Any) → BeautifulSoup

Get a URL and return parsed HTML.

Parameters:

• url – URL to fetch

• **kwargs – Additional arguments passed to get()

Returns:

BeautifulSoup object

Raises:

NetworkError – If request fails