tessl/pypi-arctic
AHL Research Versioned TimeSeries and Tick store for high-performance financial data storage and analysis
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
arctic-connection.mddocs/
Arctic Connection Management
Core functionality for connecting to MongoDB, managing libraries, and handling authentication. Provides the main entry point for all Arctic operations including library lifecycle management and quota control.
Capabilities
Arctic Class
Main Arctic database connection and library manager providing centralized access to MongoDB-backed time series storage.
class Arctic:
"""
Top-level Arctic connection managing MongoDB databases and libraries.
Each Arctic instance owns arctic_<user> databases in MongoDB, containing
one or more ArcticLibrary instances with implementation-specific functionality.
"""
def __init__(self, mongo_host, app_name='arctic', allow_secondary=False,
socketTimeoutMS=600000, connectTimeoutMS=2000,
serverSelectionTimeoutMS=30000, **kwargs):
"""
Construct Arctic datastore connection.
Parameters:
- mongo_host: MongoDB hostname, alias, or pymongo connection
- app_name: Application name for authentication (default: 'arctic')
- allow_secondary: Allow reads from secondary cluster members
- socketTimeoutMS: Socket timeout in milliseconds (default: 10 minutes)
- connectTimeoutMS: Connection timeout in milliseconds (default: 2 seconds)
- serverSelectionTimeoutMS: Server selection timeout (default: 30 seconds)
- **kwargs: Additional PyMongo MongoClient arguments (SSL, etc.)
"""Library Management
Operations for creating, listing, and managing Arctic libraries within the MongoDB cluster.
def list_libraries(self, newer_than_secs=None):
"""
List all available libraries.
Parameters:
- newer_than_secs: Only return libraries modified within this time period
Returns:
List of library names
"""
def library_exists(self, library):
"""
Check if library exists.
Parameters:
- library: Library name to check
Returns:
bool: True if library exists
"""
def initialize_library(self, library, lib_type=VERSION_STORE, **kwargs):
"""
Create new Arctic library with specified type.
Parameters:
- library: Library name to create
- lib_type: Storage engine type (VERSION_STORE, TICK_STORE, CHUNK_STORE, etc.)
- **kwargs: Library-specific initialization parameters
Raises:
- ArcticException: If library already exists or initialization fails
"""
def get_library(self, library):
"""
Get ArcticLibraryBinding instance for library operations.
Parameters:
- library: Library name
Returns:
ArcticLibraryBinding: Library interface
Raises:
- LibraryNotFoundException: If library doesn't exist
"""
def delete_library(self, library):
"""
Delete library and all its data permanently.
Parameters:
- library: Library name to delete
Raises:
- LibraryNotFoundException: If library doesn't exist
"""
def rename_library(self, from_lib, to_lib):
"""
Rename existing library.
Parameters:
- from_lib: Current library name
- to_lib: New library name
Raises:
- LibraryNotFoundException: If source library doesn't exist
- ArcticException: If target library already exists
"""
def get_library_type(self, lib):
"""
Get the storage engine type of specified library.
Parameters:
- lib: Library name
Returns:
str: Library type (VERSION_STORE, TICK_STORE, etc.)
"""Quota Management
Storage quota management for controlling disk usage and preventing runaway storage growth.
def set_quota(self, library, quota):
"""
Set storage quota in bytes for library.
Parameters:
- library: Library name
- quota: Quota in bytes (0 = unlimited)
"""
def get_quota(self, library):
"""
Get current quota for library.
Parameters:
- library: Library name
Returns:
int: Current quota in bytes (0 = unlimited)
"""
def check_quota(self, library):
"""
Check quota status and usage for library.
Parameters:
- library: Library name
Returns:
dict: Quota information including usage and limits
"""Connection Management
Methods for managing MongoDB connections and caching behavior.
def reset(self):
"""
Reset MongoDB connections.
Useful after process forking to ensure clean connection state.
"""
def is_caching_enabled(self):
"""
Check if library metadata caching is enabled.
Returns:
bool: True if caching is active
"""
def reload_cache(self):
"""
Reload the library metadata cache.
Forces refresh of cached library information.
"""Special Methods
Convenience methods for library access and Python integration.
def __getitem__(self, key):
"""
Access library using bracket notation: arctic[library_name].
Parameters:
- key: Library name
Returns:
ArcticLibraryBinding: Library interface
"""Library Type Registration
Register custom storage engine types with Arctic.
def register_library_type(name, type_):
"""
Register a custom Arctic library type handler.
Parameters:
- name: Unique name for the library type
- type_: Library class implementing the storage engine interface
Raises:
- ArcticException: If library type name already registered
Example:
register_library_type('CUSTOM_STORE', MyCustomStore)
"""ArcticLibraryBinding Class
Library-specific interface providing quota management, metadata operations, and store access.
class ArcticLibraryBinding:
"""
Library-specific interface for Arctic operations.
Provides access to library metadata, quota management, and the underlying
storage implementation.
"""
def get_name(self):
"""
Get full qualified library name.
Returns:
str: Complete library name including database prefix
"""
def get_top_level_collection(self):
"""
Get MongoDB collection reference for this library.
Returns:
pymongo.Collection: MongoDB collection object
"""
def get_library_type(self):
"""
Get library storage engine type.
Returns:
str: Library type string
"""
def set_library_type(self, lib_type):
"""
Set library storage engine type.
Parameters:
- lib_type: New library type
"""
def set_quota(self, quota_bytes):
"""
Set storage quota for this library.
Parameters:
- quota_bytes: Quota in bytes (0 = unlimited)
"""
def get_quota(self):
"""
Get current quota setting.
Returns:
int: Current quota in bytes
"""
def check_quota(self):
"""
Check quota usage status.
Returns:
dict: Quota status information
"""
def get_library_metadata(self, field):
"""
Get library metadata field value.
Parameters:
- field: Metadata field name
Returns:
Value of metadata field
"""
def set_library_metadata(self, field, value):
"""
Set library metadata field.
Parameters:
- field: Metadata field name
- value: Value to set
"""
def reset_auth(self):
"""
Reset authentication for this library.
Forces re-authentication on next operation.
"""Usage Examples
Basic Connection and Library Setup
from arctic import Arctic, VERSION_STORE, TICK_STORE
# Connect to Arctic with custom timeouts
arctic_conn = Arctic(
'mongodb://user:pass@host:27017',
app_name='my_trading_app',
socketTimeoutMS=30000,
connectTimeoutMS=5000
)
# List existing libraries
libraries = arctic_conn.list_libraries()
print(f"Available libraries: {libraries}")
# Create new libraries for different data types
arctic_conn.initialize_library('daily_prices', VERSION_STORE)
arctic_conn.initialize_library('tick_data', TICK_STORE)
# Access libraries
price_lib = arctic_conn['daily_prices']
tick_lib = arctic_conn.get_library('tick_data')Quota Management
# Set 10GB quota on library
arctic_conn.set_quota('daily_prices', 10 * 1024**3)
# Check quota status
quota_info = arctic_conn.check_quota('daily_prices')
print(f"Quota usage: {quota_info}")
# Get current quota setting
current_quota = arctic_conn.get_quota('daily_prices')
print(f"Current quota: {current_quota} bytes")Library Metadata Management
# Access library binding
lib = arctic_conn['daily_prices']
# Set custom metadata
lib.set_library_metadata('data_source', 'Bloomberg')
lib.set_library_metadata('update_frequency', 'daily')
# Retrieve metadata
source = lib.get_library_metadata('data_source')
frequency = lib.get_library_metadata('update_frequency')
print(f"Data source: {source}, Update frequency: {frequency}")Install with Tessl CLI
npx tessl i tessl/pypi-arctic