tessl/pypi-pyairtable

Python client for the Airtable API providing comprehensive database operations, ORM functionality, enterprise features, and testing utilities

—

Pending

Overview

Eval results

Files

Enterprise Features

Name: tessl/pypi-pyairtable
Author: tessl

Enterprise account management including user administration, audit logging, workspace management, and organizational controls. These features require an Enterprise billing plan and provide advanced administrative capabilities.

Capabilities

Enterprise Account Management

Core enterprise account operations including metadata retrieval, user management, and organizational structure access.

class Enterprise:
    def __init__(self, api: object, enterprise_account_id: str):
        """
        Initialize Enterprise instance.
        
        Parameters:
        - api: Api instance
        - enterprise_account_id: Enterprise account ID
        """
    
    def info(self, *, aggregated: bool = False, descendants: bool = False) -> object:
        """
        Get enterprise account information.
        
        Parameters:
        - aggregated: Include aggregated values across enterprise
        - descendants: Include information about descendant organizations
        
        Returns:
        EnterpriseInfo object with account metadata
        """
    
    def create_descendant(self, name: str) -> 'Enterprise':
        """
        Create descendant enterprise account.
        
        Parameters:
        - name: Name for the new descendant account
        
        Returns:
        Enterprise instance for created account
        """

class Workspace:
    def __init__(self, api: object, workspace_id: str):
        """
        Initialize Workspace instance.
        
        Parameters:
        - api: Api instance  
        - workspace_id: Workspace ID (starts with 'wsp')
        """
    
    def create_base(self, name: str, tables: list[dict]) -> object:
        """
        Create base in workspace.
        
        Parameters:
        - name: Base name
        - tables: List of table schema dictionaries
        
        Returns:
        Base instance for created base
        """
    
    def collaborators(self, *, force: bool = False) -> object:
        """
        Get workspace collaborators and metadata.
        
        Parameters:
        - force: Force refresh cached data
        
        Returns:
        WorkspaceCollaborators object
        """
    
    def bases(self) -> list[object]:
        """
        Get all bases within workspace.
        
        Returns:
        List of Base instances
        """
    
    @property
    def name(self) -> str:
        """Workspace name."""
    
    def delete(self) -> None:
        """Delete workspace."""
    
    def move_base(self, base: Union[str, object], target: Union[str, 'Workspace'], 
                  index: Optional[int] = None) -> None:
        """
        Move base to different workspace.
        
        Parameters:
        - base: Base ID or Base instance to move
        - target: Target workspace ID or Workspace instance
        - index: Position in target workspace (optional)
        """

User Management

Comprehensive user administration including user retrieval, access control, and account lifecycle management.

def user(self, id_or_email: str, *, 
         collaborations: bool = True,
         aggregated: bool = False, 
         descendants: bool = False) -> object:
    """
    Get single user information.
    
    Parameters:
    - id_or_email: User ID (starts with 'usr') or email address
    - collaborations: Include collaboration data
    - aggregated: Include aggregated values across enterprise
    - descendants: Include information per descendant enterprise
    
    Returns:
    UserInfo object with user details
    """

def users(self, ids_or_emails: list[str], *,
          collaborations: bool = True,
          aggregated: bool = False,
          descendants: bool = False) -> list[object]:
    """
    Get multiple users' information.
    
    Parameters:
    - ids_or_emails: List of user IDs or email addresses (max 100)
    - collaborations: Include collaboration data
    - aggregated: Include aggregated values
    - descendants: Include descendant enterprise information
    
    Returns:
    List of UserInfo objects
    """

def remove_user(self, user_id: str, replacement: Optional[str] = None, *,
                descendants: bool = False) -> object:
    """
    Remove user from enterprise and all associated workspaces/bases.
    
    Parameters:
    - user_id: User ID to remove
    - replacement: Replacement owner for sole-owned workspaces
    - descendants: Also remove from descendant enterprises
    
    Returns:
    UserRemoved object with removal details
    """

def claim_users(self, users: dict[str, str]) -> object:
    """
    Manage user membership status (managed vs unmanaged).
    
    Parameters:
    - users: Dict mapping user IDs/emails to 'managed' or 'unmanaged'
    
    Returns:
    ManageUsersResponse object with operation results
    """

def delete_users(self, emails: list[str]) -> object:
    """
    Delete multiple users by email address.
    
    Parameters:
    - emails: List of email addresses to delete
    
    Returns:
    DeleteUsersResponse object with deletion results
    """

def grant_admin(self, *users: Union[str, object]) -> object:
    """
    Grant admin access to users.
    
    Parameters:
    - users: User IDs, emails, or UserInfo objects
    
    Returns:
    ManageUsersResponse object
    """

def revoke_admin(self, *users: Union[str, object]) -> object:
    """
    Revoke admin access from users.
    
    Parameters:
    - users: User IDs, emails, or UserInfo objects
    
    Returns:
    ManageUsersResponse object
    """

Group Management

User group operations for organizing users and managing permissions at scale.

def group(self, group_id: str, collaborations: bool = True) -> object:
    """
    Get user group information.
    
    Parameters:
    - group_id: Group ID (starts with 'grp')
    - collaborations: Include collaboration data
    
    Returns:
    UserGroup object with group details
    """

def move_groups(self, group_ids: list[str], 
                target: Union[str, 'Enterprise']) -> object:
    """
    Move user groups to different enterprise account.
    
    Parameters:
    - group_ids: List of group IDs to move
    - target: Target enterprise ID or Enterprise instance
    
    Returns:
    MoveGroupsResponse object with move results
    """

def move_workspaces(self, workspace_ids: list[str],
                    target: Union[str, 'Enterprise']) -> object:
    """
    Move workspaces to different enterprise account.
    
    Parameters:
    - workspace_ids: List of workspace IDs to move  
    - target: Target enterprise ID or Enterprise instance
    
    Returns:
    MoveWorkspacesResponse object with move results
    """

Audit Logging

Comprehensive audit trail functionality for tracking all enterprise activities and changes.

def audit_log(self, *,
              page_size: Optional[int] = None,
              page_limit: Optional[int] = None,
              sort_asc: Optional[bool] = False,
              previous: Optional[str] = None,
              next: Optional[str] = None,
              start_time: Optional[Union[str, object]] = None,
              end_time: Optional[Union[str, object]] = None,
              user_id: Optional[Union[str, list[str]]] = None,
              event_type: Optional[Union[str, list[str]]] = None,
              model_id: Optional[Union[str, list[str]]] = None,
              category: Optional[Union[str, list[str]]] = None) -> Iterator[object]:
    """
    Retrieve audit log events with filtering and pagination.
    
    Parameters:
    - page_size: Events per page (max 100)
    - page_limit: Maximum pages to retrieve
    - sort_asc: Sort ascending (earliest first) vs descending (latest first)
    - previous: Pagination cursor for previous page
    - next: Pagination cursor for next page
    - start_time: Earliest timestamp to retrieve (inclusive)
    - end_time: Latest timestamp to retrieve (inclusive)
    - user_id: Filter by originating user ID(s) (max 100)
    - event_type: Filter by event type(s) (max 100)
    - model_id: Filter by model ID(s) that were affected (max 100)
    - category: Filter by event category/categories
    
    Yields:
    AuditLogResponse objects containing events and pagination info
    """

Usage Examples

Enterprise Account Setup

from pyairtable import Api

# Initialize API with enterprise access
api = Api('your_enterprise_access_token')

# Get enterprise instance
enterprise = api.enterprise('entYourEnterpriseId')

# Get enterprise information
info = enterprise.info(aggregated=True, descendants=True)
print(f"Enterprise: {info.name}")
print(f"Total users: {info.user_count}")
print(f"Total workspaces: {len(info.workspace_ids)}")

# List descendant enterprises
for descendant in info.descendants or []:
    print(f"Descendant: {descendant.name} ({descendant.id})")

User Management Operations

# Get user information
user = enterprise.user('user@company.com', collaborations=True)
print(f"User: {user.name} ({user.email})")
print(f"Admin: {user.is_admin}")
print(f"Workspaces: {len(user.collaborations.workspaces)}")

# Get multiple users efficiently
user_emails = ['user1@company.com', 'user2@company.com', 'user3@company.com']
users = enterprise.users(user_emails)

for user in users:
    print(f"{user.name}: {user.state}")  # managed/unmanaged

# Grant admin access
enterprise.grant_admin('user1@company.com', 'usrAnotherUserId')

# Manage user membership
enterprise.claim_users({
    'new_user@company.com': 'managed',
    'contractor@company.com': 'unmanaged'
})

# Remove user from enterprise
removal_result = enterprise.remove_user(
    'departing_user@company.com',
    replacement='replacement_owner@company.com'
)
print(f"Removed as admin: {removal_result.was_user_removed_as_admin}")

Workspace Management

# Get workspace
workspace = api.workspace('wspmhESAta6clCCwF')

# Get workspace details
collaborators = workspace.collaborators()
print(f"Workspace: {collaborators.name}")
print(f"Collaborators: {len(collaborators.collaborators)}")

# List bases in workspace
bases = workspace.bases()
for base in bases:
    print(f"Base: {base.name} ({base.id})")

# Create new base in workspace
new_base = workspace.create_base(
    name='New Project Base',
    tables=[
        {
            'name': 'Tasks',
            'fields': [
                {'name': 'Name', 'type': 'singleLineText'},
                {'name': 'Status', 'type': 'singleSelect', 
                 'options': {'choices': [{'name': 'Todo'}, {'name': 'Done'}]}},
                {'name': 'Due Date', 'type': 'date'}
            ]
        }
    ]
)
print(f"Created base: {new_base.id}")

# Move base between workspaces
source_workspace = api.workspace('wspmhESAta6clCCwF')
target_workspace = api.workspace('wspAnotherWorkspace')

source_workspace.move_base(
    base=new_base,
    target=target_workspace,
    index=0  # Move to first position
)

Audit Log Monitoring

from datetime import datetime, timedelta

# Get recent audit events
recent_events = []
for page in enterprise.audit_log(
    sort_asc=False,  # Latest first
    page_size=100,
    page_limit=5,    # First 5 pages
    start_time=datetime.now() - timedelta(days=7)  # Last week
):
    recent_events.extend(page.events)

print(f"Found {len(recent_events)} events in last week")

# Filter by specific event types
security_events = enterprise.audit_log(
    event_type=['user.created', 'user.deleted', 'admin.granted', 'admin.revoked'],
    start_time=datetime.now() - timedelta(days=30)
)

for page in security_events:
    for event in page.events:
        print(f"{event.event_time}: {event.event_type} by {event.originating_user_id}")

# Monitor specific users' activities
user_activities = enterprise.audit_log(
    user_id=['usr1234567890abcd', 'usr0987654321fedc'],
    category=['data'],  # Data-related events only
    sort_asc=True
)

for page in user_activities:
    for event in page.events:
        print(f"{event.originating_user_id}: {event.event_type} on {event.model_id}")

Enterprise Organization Management

# Create descendant enterprise (for Enterprise Hub customers)
child_enterprise = enterprise.create_descendant('Regional Office East')
print(f"Created descendant: {child_enterprise.id}")

# Move user groups between enterprises
group_move_result = enterprise.move_groups(
    group_ids=['grp1234567890abcd', 'grp0987654321fedc'],
    target=child_enterprise
)

print(f"Moved groups: {len(group_move_result.moved_groups)}")
if group_move_result.errors:
    for error in group_move_result.errors:
        print(f"Error moving {error.id}: {error.message}")

# Move workspaces between enterprises
workspace_move_result = enterprise.move_workspaces(
    workspace_ids=['wspmhESAta6clCCwF'],
    target='entAnotherEnterpriseId'
)

print(f"Moved workspaces: {len(workspace_move_result.moved_workspaces)}")

Advanced User Group Management

# Get group information
group = enterprise.group('grp1234567890abcd', collaborations=True)
print(f"Group: {group.name}")
print(f"Members: {len(group.user_ids)}")

# Access group collaborations
for workspace_collab in group.collaborations.workspaces:
    print(f"Workspace {workspace_collab.workspace_id}: {workspace_collab.permission_level}")

for base_collab in group.collaborations.bases:
    print(f"Base {base_collab.base_id}: {base_collab.permission_level}")

Bulk User Operations

# Process all enterprise users
all_user_ids = enterprise.info().user_ids
inactive_users = []

# Check users in batches (API limit: 100 per request)
for i in range(0, len(all_user_ids), 100):
    batch = all_user_ids[i:i+100]
    users = enterprise.users(batch, collaborations=True)
    
    for user in users:
        # Find users with no recent activity
        if not user.collaborations.bases and not user.collaborations.workspaces:
            inactive_users.append(user.email)

print(f"Found {len(inactive_users)} inactive users")

# Bulk manage inactive users
if inactive_users:
    # Convert to unmanaged status
    user_changes = {email: 'unmanaged' for email in inactive_users[:50]}  # First 50
    result = enterprise.claim_users(user_changes)
    
    if result.errors:
        for error in result.errors:
            print(f"Error processing {error.email}: {error.message}")

Install with Tessl CLI