tessl/pypi-minio
MinIO Python SDK for Amazon S3 Compatible Cloud Storage
—
Pending
admin-operations.mddocs/
Administrative Operations
MinIO server administration operations using the MinioAdmin class. These operations provide comprehensive control over MinIO server instances including user management, policy configuration, service accounts, server configuration, and system monitoring.
Capabilities
Service Management
Control MinIO server lifecycle including restart, stop, update, and information retrieval.
class MinioAdmin:
"""MinIO administration client."""
def __init__(
self,
endpoint: str,
credentials: Provider,
region: str = "",
secure: bool = True,
cert_check: bool = True,
http_client: urllib3.PoolManager | None = None
) -> None:
"""
Initialize MinIO admin client.
Args:
endpoint: MinIO server endpoint (hostname:port)
credentials: Credential provider instance
region: AWS region (optional for MinIO)
secure: Use HTTPS (default: True)
cert_check: Verify SSL certificates (default: True)
http_client: Custom HTTP client (optional)
"""
def service_restart(self) -> str:
"""
Restart MinIO service.
Returns:
Success message from server
Raises:
MinioAdminException: If restart fails
"""
def service_stop(self) -> str:
"""
Stop MinIO service.
Returns:
Success message from server
Raises:
MinioAdminException: If stop fails
"""
def info(self) -> str:
"""
Get server information including version, uptime, and status.
Returns:
JSON string containing server information
Raises:
MinioAdminException: If info retrieval fails
"""
def update(self) -> str:
"""
Update MinIO server to latest version.
Returns:
Update status message
Raises:
MinioAdminException: If update fails
"""User Management
Create, modify, and manage MinIO users with access control and status management.
def user_add(self, access_key: str, secret_key: str) -> str:
"""
Add a new user to MinIO.
Args:
access_key: User's access key (username)
secret_key: User's secret key (password)
Returns:
Success message
Raises:
MinioAdminException: If user creation fails
"""
def user_remove(self, access_key: str) -> str:
"""
Remove a user from MinIO.
Args:
access_key: User's access key to remove
Returns:
Success message
Raises:
MinioAdminException: If user removal fails
"""
def user_list(self) -> str:
"""
List all users in MinIO.
Returns:
JSON string containing user list
Raises:
MinioAdminException: If listing fails
"""
def user_info(self, access_key: str) -> str:
"""
Get detailed information about a specific user.
Args:
access_key: User's access key
Returns:
JSON string containing user information
Raises:
MinioAdminException: If user info retrieval fails
"""
def user_enable(self, access_key: str) -> str:
"""
Enable a disabled user account.
Args:
access_key: User's access key to enable
Returns:
Success message
Raises:
MinioAdminException: If enable operation fails
"""
def user_disable(self, access_key: str) -> str:
"""
Disable a user account without deletion.
Args:
access_key: User's access key to disable
Returns:
Success message
Raises:
MinioAdminException: If disable operation fails
"""Group Management
Manage user groups for simplified permission management and organizational structure.
def group_add(self, group_name: str, members: str) -> str:
"""
Add a new group with specified members.
Args:
group_name: Name of the group to create
members: Comma-separated string of user access keys to add to group
Returns:
Success message
Raises:
MinioAdminException: If group creation fails
"""
def group_remove(self, group_name: str, members: str | None = None) -> str:
"""
Remove group or remove specific members from group.
Args:
group_name: Name of the group
members: Comma-separated string of members to remove (None removes entire group)
Returns:
Success message
Raises:
MinioAdminException: If group removal fails
"""
def group_list(self) -> str:
"""
List all groups in MinIO.
Returns:
JSON string containing group list
Raises:
MinioAdminException: If listing fails
"""
def group_info(self, group_name: str) -> str:
"""
Get detailed information about a specific group.
Args:
group_name: Name of the group
Returns:
JSON string containing group information and members
Raises:
MinioAdminException: If group info retrieval fails
"""
def group_enable(self, group_name: str) -> str:
"""
Enable a disabled group.
Args:
group_name: Name of the group to enable
Returns:
Success message
Raises:
MinioAdminException: If enable operation fails
"""
def group_disable(self, group_name: str) -> str:
"""
Disable a group without deletion.
Args:
group_name: Name of the group to disable
Returns:
Success message
Raises:
MinioAdminException: If disable operation fails
"""Policy Management
Manage access policies for fine-grained permission control across users and groups.
def policy_add(self, policy_name: str, policy_file: str) -> str:
"""
Add a new policy from file or JSON string.
Args:
policy_name: Name for the new policy
policy_file: Path to policy file or JSON policy string
Returns:
Success message
Raises:
MinioAdminException: If policy creation fails
"""
def policy_remove(self, policy_name: str) -> str:
"""
Remove an existing policy.
Args:
policy_name: Name of the policy to remove
Returns:
Success message
Raises:
MinioAdminException: If policy removal fails
"""
def policy_list(self) -> str:
"""
List all policies in MinIO.
Returns:
JSON string containing policy list
Raises:
MinioAdminException: If listing fails
"""
def policy_info(self, policy_name: str) -> str:
"""
Get detailed information about a specific policy.
Args:
policy_name: Name of the policy
Returns:
JSON string containing policy definition
Raises:
MinioAdminException: If policy info retrieval fails
"""
def policy_set(self, policy_name: str, user: str, is_group: bool = False) -> str:
"""
Attach a policy to a user or group.
Args:
policy_name: Name of the policy to attach
user: User access key or group name
is_group: True if user parameter is a group name
Returns:
Success message
Raises:
MinioAdminException: If policy attachment fails
"""
def policy_unset(self, policy_name: str, user: str, is_group: bool = False) -> str:
"""
Detach a policy from a user or group.
Args:
policy_name: Name of the policy to detach
user: User access key or group name
is_group: True if user parameter is a group name
Returns:
Success message
Raises:
MinioAdminException: If policy detachment fails
"""Service Accounts
Manage service accounts for application access with limited scope and temporary credentials.
def add_service_account(
self,
access_key: str,
secret_key: str,
user: str,
policy: str | None = None,
name: str | None = None,
description: str | None = None,
expiry: datetime.datetime | None = None
) -> str:
"""
Create a new service account.
Args:
access_key: Service account access key
secret_key: Service account secret key
user: Parent user for the service account
policy: JSON policy string (optional)
name: Friendly name for service account (optional)
description: Description of service account (optional)
expiry: Expiration date for service account (optional)
Returns:
JSON string containing service account details
Raises:
MinioAdminException: If service account creation fails
"""
def update_service_account(
self,
access_key: str,
secret_key: str | None = None,
policy: str | None = None,
name: str | None = None,
description: str | None = None,
expiry: datetime.datetime | None = None
) -> str:
"""
Update an existing service account.
Args:
access_key: Service account access key to update
secret_key: New secret key (optional)
policy: New JSON policy string (optional)
name: New friendly name (optional)
description: New description (optional)
expiry: New expiration date (optional)
Returns:
Success message
Raises:
MinioAdminException: If service account update fails
"""
def delete_service_account(self, access_key: str) -> str:
"""
Delete a service account.
Args:
access_key: Service account access key to delete
Returns:
Success message
Raises:
MinioAdminException: If service account deletion fails
"""
def get_service_account(self, access_key: str) -> str:
"""
Get information about a service account.
Args:
access_key: Service account access key
Returns:
JSON string containing service account details
Raises:
MinioAdminException: If service account retrieval fails
"""
def list_service_account(self, user: str | None = None) -> str:
"""
List service accounts for a user or all service accounts.
Args:
user: User to list service accounts for (None for all)
Returns:
JSON string containing service account list
Raises:
MinioAdminException: If listing fails
"""Configuration Management
Manage MinIO server configuration including system settings and feature toggles.
def config_get(self, key: str) -> str:
"""
Get configuration value for a specific key.
Args:
key: Configuration key to retrieve
Returns:
Configuration value as string
Raises:
MinioAdminException: If configuration retrieval fails
"""
def config_set(self, key: str, value: str, **kwargs) -> str:
"""
Set configuration value for a specific key.
Args:
key: Configuration key to set
value: Configuration value to set
**kwargs: Additional configuration parameters
Returns:
Success message
Raises:
MinioAdminException: If configuration setting fails
"""
def config_reset(self, key: str, name: str) -> str:
"""
Reset configuration to default values.
Args:
key: Configuration section to reset
name: Specific configuration name to reset
Returns:
Success message
Raises:
MinioAdminException: If configuration reset fails
"""
def config_history(self) -> str:
"""
Get configuration change history.
Returns:
JSON string containing configuration history
Raises:
MinioAdminException: If history retrieval fails
"""
def config_restore(self, restore_id: str) -> str:
"""
Restore configuration from historical snapshot.
Args:
restore_id: ID of configuration snapshot to restore
Returns:
Success message
Raises:
MinioAdminException: If configuration restore fails
"""Monitoring and Diagnostics
Monitor server performance, health, and operational metrics.
def account_info(self, prefix_usage: bool = False) -> str:
"""
Get account usage information and statistics.
Args:
prefix_usage: Include prefix-based usage statistics
Returns:
JSON string containing account information
Raises:
MinioAdminException: If account info retrieval fails
"""
def bucket_quota_set(self, bucket: str, size: int) -> str:
"""
Set storage quota for a bucket.
Args:
bucket: Name of the bucket
size: Maximum size in bytes (0 to remove quota)
Returns:
Success message
Raises:
MinioAdminException: If quota setting fails
"""
def bucket_quota_get(self, bucket: str) -> str:
"""
Get storage quota for a bucket.
Args:
bucket: Name of the bucket
Returns:
JSON string containing quota information
Raises:
MinioAdminException: If quota retrieval fails
"""
def top_locks(self) -> str:
"""
Get information about top resource locks on the server.
Returns:
JSON string containing lock information
Raises:
MinioAdminException: If lock info retrieval fails
"""
def profile_start(self, profilers: tuple[str] = ()) -> str:
"""
Start system profiling on MinIO server.
Args:
profilers: Tuple of profiler types to enable
Returns:
Success message from server
Raises:
MinioAdminException: If profiling start fails
"""
def get_data_usage_info(self) -> str:
"""
Get data usage information for the MinIO deployment.
Returns:
JSON string containing data usage statistics
Raises:
MinioAdminException: If data usage retrieval fails
"""
def bucket_quota_clear(self, bucket: str) -> str:
"""
Clear storage quota for a bucket.
Args:
bucket: Name of the bucket
Returns:
Success message
Raises:
MinioAdminException: If quota clearing fails
"""Site Replication
Manage multi-site replication configuration for disaster recovery and data synchronization across MinIO deployments.
def add_site_replication(self, peer_sites: list[PeerSite]) -> str:
"""
Add peer sites to site replication configuration.
Args:
peer_sites: List of peer site configurations
Returns:
Success message with replication details
Raises:
MinioAdminException: If site replication setup fails
"""
def get_site_replication_info(self) -> str:
"""
Get site replication information and status.
Returns:
JSON string containing site replication configuration
Raises:
MinioAdminException: If info retrieval fails
"""
def get_site_replication_status(
self,
options: SiteReplicationStatusOptions
) -> str:
"""
Get detailed site replication status.
Args:
options: Status query options and filters
Returns:
JSON string containing replication status details
Raises:
MinioAdminException: If status retrieval fails
"""
def edit_site_replication(self, peer_info: PeerInfo) -> str:
"""
Edit site replication configuration with peer information.
Args:
peer_info: Updated peer configuration information
Returns:
Success message
Raises:
MinioAdminException: If replication edit fails
"""
def remove_site_replication(
self,
sites: str | None = None,
all_sites: bool = False
) -> str:
"""
Remove sites from site replication configuration.
Args:
sites: Comma-separated list of sites to remove
all_sites: Remove all sites from replication
Returns:
Success message
Raises:
MinioAdminException: If site removal fails
"""Encryption and Security
Manage KMS keys and encryption settings for enhanced security.
def kms_key_create(self, key: str) -> str:
"""
Create a new KMS encryption key.
Args:
key: Key identifier or configuration
Returns:
Success message with key details
Raises:
MinioAdminException: If key creation fails
"""
def kms_key_status(self, key: str) -> str:
"""
Get status information for a KMS key.
Args:
key: Key identifier
Returns:
JSON string containing key status
Raises:
MinioAdminException: If key status retrieval fails
"""Advanced Policy Management
Extended policy management including LDAP integration and bulk operations.
def attach_policy_ldap(
self,
policies: list[str],
user: str | None = None,
group: str | None = None
) -> str:
"""
Attach policies for LDAP user or group.
Args:
policies: List of policy names to attach
user: LDAP user to attach policies to
group: LDAP group to attach policies to
Returns:
Success message
Raises:
MinioAdminException: If policy attachment fails
"""
def detach_policy_ldap(
self,
policies: list[str],
user: str | None = None,
group: str | None = None
) -> str:
"""
Detach policies from LDAP user or group.
Args:
policies: List of policy names to detach
user: LDAP user to detach policies from
group: LDAP group to detach policies from
Returns:
Success message
Raises:
MinioAdminException: If policy detachment fails
"""
def list_access_keys_ldap(
self,
user_dn: str,
list_type: str
) -> str:
"""
List access keys for LDAP user.
Args:
user_dn: LDAP user distinguished name
list_type: Type of listing to perform
Returns:
JSON string containing access keys
Raises:
MinioAdminException: If listing fails
"""
def list_access_keys_ldap_bulk(
self,
users: list[str],
list_type: str,
all_users: bool
) -> str:
"""
List access keys for multiple LDAP users or all users.
Args:
users: List of user DNs to list keys for
list_type: Type of listing to perform
all_users: List keys for all users
Returns:
JSON string containing access keys
Raises:
MinioAdminException: If bulk listing fails
"""
def attach_policy(
self,
policies: list[str],
user: str | None = None,
group: str | None = None
) -> str:
"""
Attach builtin policies to user or group.
Args:
policies: List of policy names to attach
user: User to attach policies to
group: Group to attach policies to
Returns:
Success message
Raises:
MinioAdminException: If policy attachment fails
"""
def detach_policy(
self,
policies: list[str],
user: str | None = None,
group: str | None = None
) -> str:
"""
Detach builtin policies from user or group.
Args:
policies: List of policy names to detach
user: User to detach policies from
group: Group to detach policies from
Returns:
Success message
Raises:
MinioAdminException: If policy detachment fails
"""
def get_policy_entities(
self,
users: list[str],
groups: list[str],
policies: list[str]
) -> str:
"""
Get policy entities information.
Args:
users: List of users to query
groups: List of groups to query
policies: List of policies to query
Returns:
JSON string containing policy entities
Raises:
MinioAdminException: If entity retrieval fails
"""Types
Admin Exception Types
class MinioAdminException(Exception):
"""Exception for MinIO admin operations."""
def __init__(self, code: int, body: str) -> None: ...
code: int
body: strSite Replication Types
class PeerSite:
"""Represents a cluster/site to be added to the set of replicated sites."""
def __init__(
self,
name: str,
endpoint: str,
access_key: str,
secret_key: str
) -> None: ...
def to_dict(self) -> dict[str, str]:
"""Convert to dictionary representation."""
class PeerInfo:
"""Site replication peer information."""
def __init__(
self,
deployment_id: str,
endpoint: str,
bucket_bandwidth_limit: str,
bucket_bandwidth_set: str
) -> None: ...
deployment_id: str
endpoint: str
name: str | None
sync_status: str | None
bucket_bandwidth_limit: str
bucket_bandwidth_set: str
bucket_bandwidth_updated_at: datetime.datetime | None
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary representation."""
class SiteReplicationStatusOptions:
"""Represents site replication status options."""
def __init__(self) -> None: ...
buckets: bool
policies: bool
users: bool
groups: bool
metrics: bool
entity: str | None
entity_value: str | None
show_deleted: bool
def to_query_params(self) -> dict[str, Any]:
"""Convert to query parameters for API requests."""Usage Examples
Basic Admin Operations
from minio import MinioAdmin
from minio.credentials import StaticProvider
from minio.error import MinioAdminException
# Create admin client
admin = MinioAdmin(
"localhost:9000",
credentials=StaticProvider("admin", "password")
)
try:
# Get server information
info = admin.info()
print(f"Server info: {info}")
# List all users
users = admin.user_list()
print(f"Users: {users}")
# Add a new user
result = admin.user_add("newuser", "newpassword123")
print(f"User created: {result}")
except MinioAdminException as e:
print(f"Admin operation failed: {e.code} - {e.body}")User and Group Management
try:
# Create users
admin.user_add("alice", "alice123")
admin.user_add("bob", "bob123")
# Create a group with members
admin.group_add("developers", "alice,bob")
# Create a policy
policy_json = """{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:PutObject"],
"Resource": ["arn:aws:s3:::dev-bucket/*"]
}]
}"""
admin.policy_add("dev-policy", policy_json)
# Attach policy to group
admin.policy_set("dev-policy", "developers", is_group=True)
print("User management setup complete")
except MinioAdminException as e:
print(f"Setup failed: {e}")Service Account Management
import datetime
try:
# Create service account with expiry
expiry_date = datetime.datetime.now() + datetime.timedelta(days=30)
service_account = admin.add_service_account(
access_key="svc-app1",
secret_key="svc-password123",
user="alice",
name="Application 1 Service Account",
description="Service account for application 1 access",
expiry=expiry_date
)
print(f"Service account created: {service_account}")
# List service accounts for user
accounts = admin.list_service_account("alice")
print(f"Alice's service accounts: {accounts}")
# Update service account
admin.update_service_account(
access_key="svc-app1",
description="Updated description for app 1"
)
except MinioAdminException as e:
print(f"Service account operation failed: {e}")Configuration and Monitoring
try:
# Get current configuration
config = admin.config_get("api")
print(f"API configuration: {config}")
# Set bucket quota
admin.bucket_quota_set("user-uploads", 1024*1024*1024) # 1GB
quota = admin.bucket_quota_get("user-uploads")
print(f"Bucket quota: {quota}")
# Get account usage info
usage = admin.account_info(prefix_usage=True)
print(f"Account usage: {usage}")
# Check top locks
locks = admin.top_locks()
print(f"Top locks: {locks}")
except MinioAdminException as e:
print(f"Monitoring operation failed: {e}")Server Lifecycle Management
try:
# Get server info before maintenance
info = admin.info()
print(f"Server status: {info}")
# Perform server update
update_result = admin.update()
print(f"Update result: {update_result}")
# Restart server to apply updates
restart_result = admin.service_restart()
print(f"Restart result: {restart_result}")
except MinioAdminException as e:
print(f"Server management failed: {e}")
# Emergency stop if needed
try:
stop_result = admin.service_stop()
print(f"Emergency stop: {stop_result}")
except MinioAdminException as stop_e:
print(f"Stop failed: {stop_e}")Install with Tessl CLI
npx tessl i tessl/pypi-minio