PyAirtable

from pyairtable import Api, Base, Table

from pyairtable import Enterprise, Workspace, retry_strategy
from pyairtable.api import retrying  # For retry strategy types
from pyairtable.api.types import UserAndScopesDict, BaseSchema  # For type annotations
from pyairtable.formulas import Formula, AND, OR, match
from pyairtable.orm import Model, fields
from pyairtable.testing import MockAirtable, fake_record

from pyairtable import Api

# Initialize API connection
api = Api('your_access_token')

# Get a table
table = api.table('app1234567890abcde', 'My Table')

# Create a record
new_record = table.create({'Name': 'John Doe', 'Email': 'john@example.com'})
print(f"Created record: {new_record['id']}")

# Retrieve all records
records = table.all()
for record in records:
    print(f"{record['id']}: {record['fields']}")

# Update a record
updated = table.update(new_record['id'], {'Name': 'Jane Doe'})

# Delete a record
table.delete(new_record['id'])

class Api:
    def __init__(self, api_key: str, *, timeout: Optional[tuple] = None, 
                 retry_strategy: Optional[Union[bool, retrying.Retry]] = True,
                 endpoint_url: str = "https://api.airtable.com",
                 use_field_ids: bool = False): ...
    def base(self, base_id: str, validate: bool = False, force: bool = False) -> Base: ...
    def table(self, base_id: str, table_name: str, validate: bool = False, force: bool = False) -> Table: ...
    def whoami(self) -> UserAndScopesDict: ...

class Base:
    def table(self, id_or_name: str, validate: bool = False, force: bool = False) -> Table: ...
    def tables(self, force: bool = False) -> list: ...
    def schema(self, *, force: bool = False) -> BaseSchema: ...

class Table:
    def all(self, **options) -> list: ...
    def get(self, record_id: str, **options) -> dict: ...
    def create(self, fields: dict, typecast: bool = False, use_field_ids: Optional[bool] = None) -> dict: ...
    def update(self, record_id: str, fields: dict, replace: bool = False, typecast: bool = False, use_field_ids: Optional[bool] = None) -> dict: ...
    def delete(self, record_id: str) -> dict: ...

def batch_create(self, records: list, typecast: bool = False, use_field_ids: Optional[bool] = None) -> list: ...
def batch_update(self, records: list, replace: bool = False, typecast: bool = False, use_field_ids: Optional[bool] = None) -> list: ...
def batch_upsert(self, records: list, key_fields: list, replace: bool = False, typecast: bool = False, use_field_ids: Optional[bool] = None) -> dict: ...
def batch_delete(self, record_ids: list) -> list: ...
def iterate(self, **options) -> Iterator: ...
def first(self, **options) -> Optional[dict]: ...

class Formula:
    def __init__(self, value: str): ...
    def flatten(self) -> Formula: ...

def AND(*components, **fields) -> Formula: ...
def OR(*components, **fields) -> Formula: ...
def NOT(component, **fields) -> Formula: ...
def match(field_values: dict, match_any: bool = False) -> Formula: ...

# Math functions
def SUM(number, *numbers) -> Formula: ...
def AVERAGE(number, *numbers) -> Formula: ...
def MAX(number, *numbers) -> Formula: ...

# Text functions  
def CONCATENATE(text, *texts) -> Formula: ...
def FIND(string_to_find, where_to_search, start_from_position=None) -> Formula: ...

# Date functions
def DATEADD(date, number, units) -> Formula: ...
def DATETIME_FORMAT(date, output_format=None) -> Formula: ...

class Model:
    def __init__(self, **field_values): ...
    def save(self) -> object: ...
    def delete(self) -> bool: ...
    def to_record(self) -> dict: ...
    @classmethod
    def from_record(cls, record: dict) -> Model: ...
    @classmethod
    def all(cls, **options) -> list: ...

# Field types
class TextField:
    def __init__(self, field_name: str): ...

class NumberField:
    def __init__(self, field_name: str): ...

class CheckboxField:
    def __init__(self, field_name: str): ...

class Enterprise:
    def info(self, aggregated: bool = False, descendants: bool = False) -> object: ...
    def user(self, id_or_email: str, collaborations: bool = True) -> object: ...
    def users(self, ids_or_emails: list) -> list: ...
    def audit_log(self, **kwargs) -> Iterator: ...
    def remove_user(self, user_id: str, replacement: str = None) -> object: ...

class Workspace:
    def create_base(self, name: str, tables: list) -> object: ...
    def collaborators(self, force: bool = False) -> object: ...
    def delete(self) -> None: ...

class MockAirtable:
    def __init__(self, passthrough: bool = False): ...
    def add_records(self, base_id: str, table_name: str, records: list) -> list: ...
    def set_records(self, base_id: str, table_name: str, records: list) -> None: ...

def fake_record(fields: dict = None, id: str = None, **other_fields) -> dict: ...
def fake_user(value=None) -> dict: ...
def fake_attachment(url: str = "", filename: str = "") -> dict: ...

def upload_attachment(self, record_id: str, field: str, filename: str, 
                     content: Optional[bytes] = None, content_type: str = None) -> dict: ...

def comments(self, record_id: str) -> list: ...
def add_comment(self, record_id: str, text: str) -> object: ...

class Comment:
    def save(self) -> Comment: ...
    def delete(self) -> bool: ...

def webhooks(self) -> list: ...
def add_webhook(self, notify_url: str, spec: dict) -> object: ...

class Webhook:
    # Webhook properties and methods
    pass

# CLI commands available via: python -m pyairtable
# - whoami: Get current user information  
# - bases: List all available bases
# - base <id> schema: Get base schema
# - base <id> table <name> records: Get table records
# - enterprise <id> users: List enterprise users

# Record structure
RecordDict = {
    "id": str,  # Record ID starting with "rec"
    "createdTime": str,  # ISO 8601 datetime string
    "fields": dict  # Field name -> field value mapping
}

# Common field value types
FieldValue = Union[str, int, float, bool, list, dict, None]

# Attachment structure  
AttachmentDict = {
    "id": str,  # Attachment ID starting with "att"
    "url": str,  # URL to access the attachment
    "filename": str,  # Original filename
    "size": int,  # File size in bytes
    "type": str  # MIME type
}

# User/Collaborator structure
CollaboratorDict = {
    "id": str,  # User ID starting with "usr"
    "email": str,  # User email address
    "name": str  # Display name
}

# Retry strategy for API requests
TimeoutTuple = tuple[int, int]  # (connect_timeout, read_timeout)

class PyAirtableError(Exception):
    """Base class for all exceptions raised by PyAirtable."""

class CircularFormulaError(PyAirtableError, RecursionError):
    """A circular dependency was encountered when flattening nested conditions."""

class InvalidParameterError(PyAirtableError, ValueError):
    """Raised when invalid parameters are passed to all(), first(), etc."""

class MissingValueError(PyAirtableError, ValueError):
    """A required field received an empty value, either from Airtable or other code."""

class MultipleValuesError(PyAirtableError, ValueError):
    """SingleLinkField received more than one value from either Airtable or calling code."""

class ReadonlyFieldError(PyAirtableError, ValueError):
    """Attempted to set a value on a readonly field."""

class UnsavedRecordError(PyAirtableError, ValueError):
    """Attempted to perform an unsupported operation on an unsaved record."""

import requests
from pyairtable import Api
from pyairtable.exceptions import PyAirtableError, InvalidParameterError

try:
    api = Api('invalid_token')
    records = api.table('base_id', 'table_name').all()
except requests.exceptions.HTTPError as e:
    print(f"HTTP Error: {e}")
except InvalidParameterError as e:
    print(f"Invalid Parameter: {e}")
except PyAirtableError as e:
    print(f"PyAirtable Error: {e}")