Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions pyairtable/api/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union

import pyairtable.api.table
from pyairtable.exceptions import MissingRecordError
from pyairtable.models.schema import BaseCollaborators, BaseSchema, BaseShares
from pyairtable.models.webhook import (
CreateWebhook,
Expand Down Expand Up @@ -241,15 +242,15 @@ def webhooks(self) -> List[Webhook]:

def webhook(self, webhook_id: str) -> Webhook:
"""
Build a single webhook or raises ``KeyError`` if the given ID is invalid.
Build a single webhook or raises ``MissingRecordError`` if the given ID is invalid.

Airtable's API does not permit retrieving a single webhook, so this function
will call :meth:`~webhooks` and simply return one item from the list.
"""
for webhook in self.webhooks():
if webhook.id == webhook_id:
return webhook
raise KeyError(f"webhook not found: {webhook_id!r}")
raise MissingRecordError(webhook_id)

def add_webhook(
self,
Expand Down
145 changes: 142 additions & 3 deletions pyairtable/api/enterprise.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,15 +9,23 @@
List,
Literal,
Optional,
Sequence,
Union,
)

import pydantic
from typing_extensions import Self

from pyairtable.exceptions import InvalidParameterError, MissingRecordError
from pyairtable.models._base import AirtableModel, rebuild_models
from pyairtable.models.audit import AuditLogResponse
from pyairtable.models.schema import EnterpriseInfo, NestedId, UserGroup, UserInfo
from pyairtable.models.schema import (
EnterpriseInfo,
NestedId,
Package,
UserGroup,
UserInfo,
)
from pyairtable.utils import (
Url,
UrlBuilder,
Expand All @@ -29,6 +37,7 @@

if TYPE_CHECKING:
from pyairtable.api.api import Api
from pyairtable.api.base import Base
from pyairtable.api.workspace import Workspace


Expand Down Expand Up @@ -70,6 +79,15 @@ class _urls(UrlBuilder):
#: URL for creating a new workspace.
create_workspace = Url("meta/workspaces")

#: URL for listing enterprise packages.
packages = meta / "packages"

def package_install(self, package_id: str) -> Url:
"""
URL for installing a package (creating a base from a package).
"""
return self.meta / "packages" / package_id / "install"

def user(self, user_id: str) -> Url:
"""
URL for retrieving information about a single user.
Expand Down Expand Up @@ -102,9 +120,9 @@ def remove_user(self, user_id: str) -> Url:

urls = cached_property(_urls)

def __init__(self, api: "Api", workspace_id: str):
def __init__(self, api: "Api", enterprise_id: str):
self.api = api
self.id = workspace_id
self.id = enterprise_id
self._info: Optional[EnterpriseInfo] = None

@cache_unless_forced
Expand Down Expand Up @@ -539,6 +557,127 @@ def create_workspace(self, name: str) -> "Workspace":
)
return self.api.workspace(str(response["id"]))

@cache_unless_forced
def packages(
self,
*,
all_enterprises: bool = False,
) -> List[Package]:
"""
List all packages for the enterprise account.

See `List packages <https://airtable.com/developers/web/api/list-enterprise-packages>`__.

Args:
all_enterprises: If True and the enterprise account is the root
enterprise account, returns all packages across the entire
enterprise grid. Defaults to False.

Returns:
A list of Package objects representing the enterprise packages.
"""
params: Dict[str, Any] = {}
if all_enterprises:
params["shouldGetAllPackagesInGrid"] = True

response = self.api.get(self.urls.packages, params=params)
return [
Package.from_api(pkg, self.api, context=self)
for pkg in response.get("packages", [])
]

def package(self, package_id: str, *, force: bool = False) -> Package:
"""
Retrieve information about a single package by ID.

Args:
package_id: The ID of the package to retrieve.
force: If ``True``, forces a refresh of the cached package list.

Returns:
A Package object representing the enterprise package.
"""
try:
return next(
package
for package in self.packages(force=force)
if package.id == package_id
)
except StopIteration:
raise MissingRecordError(package_id)

def create_base(
self,
workspace: Union[str, "Workspace"],
name: str,
tables: Sequence[Dict[str, Any]],
) -> "Base":
"""
Create a base in the given workspace.

See https://airtable.com/developers/web/api/create-base

Args:
workspace: The ID of the workspace or a :class:`~pyairtable.Workspace` object.
name: The name to give to the new base. Does not need to be unique.
tables: A list of ``dict`` objects that conform to Airtable's
`Table model <https://airtable.com/developers/web/api/model/table-model>`__.
"""
if isinstance(workspace, str):
workspace = self.api.workspace(workspace)
return workspace.create_base(name, tables)

def create_base_from_package(
self,
workspace: Union[str, "Workspace"],
name: str,
package: Union[str, Package],
*,
description: Optional[str] = None,
release_id: Optional[str] = None,
) -> "Base":
"""
Create a base from an enterprise package template in the specified workspace.

See https://airtable.com/developers/web/api/create-base-from-package-enterprise

Args:
workspace: The ID of the workspace or a :class:`~pyairtable.Workspace` object.
name: The name for the new base.
package: Package ID (str) or Package object to install.
description: Optional description for the base.
release_id: The package release ID to install. If not provided,
attempts to use ``Package.latest_release_id`` if package is a Package object,
or calls the :meth:`Enterprise.packages` method to find it.

Returns:
The newly created Base object.

Raises:
MissingRecordError: If the specified package ID does not exist.
InvalidParameterError: If release_id cannot be determined.
"""
workspace_id = workspace if isinstance(workspace, str) else workspace.id
package_id = package if isinstance(package, str) else package.id

# Only fetch the list of packages if we need to get the latest release ID
if release_id is None:
package = self.package(package_id) if isinstance(package, str) else package
release_id = package.latest_release_id
if release_id is None:
raise InvalidParameterError("release_id is required")

payload: Dict[str, Any] = {
"name": name,
"packageReleaseId": release_id,
"workspaceId": workspace_id,
}
if description is not None:
payload["description"] = description

response = self.api.post(self.urls.package_install(package_id), json=payload)
return self.api.base(response["id"], validate=True, force=True)


class UserRemoved(AirtableModel):
"""
Expand Down
6 changes: 6 additions & 0 deletions pyairtable/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,12 @@ class InvalidParameterError(PyAirtableError, ValueError):
"""


class MissingRecordError(PyAirtableError, KeyError):
"""
A requested record was not found in Airtable.
"""


class MissingValueError(PyAirtableError, ValueError):
"""
A required field received an empty value, either from Airtable or other code.
Expand Down
22 changes: 22 additions & 0 deletions pyairtable/models/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -613,6 +613,28 @@ class AggregatedIds(AirtableModel):
workspace_ids: List[str] = _FL()


class Package(AirtableModel):
"""
Represents an enterprise package.

Returned from the `List packages <https://airtable.com/developers/web/api/list-enterprise-packages>`__ endpoint.
"""

id: str
type: str
created_by_user_id: str
created_time: datetime
description: Optional[str] = None
enterprise_account_id: Optional[str] = None
install_count: int
last_updated_by_user_id: str
last_updated_time: datetime
latest_release_id: Optional[str] = None
name: str
source_application_id: str
tagline: Optional[str] = None


class WorkspaceCollaborators(_Collaborators, url="meta/workspaces/{self.id}"):
"""
Detailed information about who can access a workspace.
Expand Down
6 changes: 5 additions & 1 deletion pyairtable/orm/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
UpdateRecordDict,
WritableFields,
)
from pyairtable.exceptions import MissingRecordError
from pyairtable.formulas import EQ, OR, RECORD_ID
from pyairtable.models import Comment
from pyairtable.orm.fields import AnyField, Field
Expand Down Expand Up @@ -466,7 +467,10 @@ def from_ids(
{obj.id: obj for obj in cls.all(formula=formula, memoize=memoize)}
)

# Ensure we return records in the same order, and raise KeyError if any are missing
# Ensure we return records in the same order, and raise if any are missing
if missing_ids := set(record_ids) - set(by_id):
raise MissingRecordError(sorted(missing_ids))

return [by_id[record_id] for record_id in record_ids]

@classmethod
Expand Down
15 changes: 15 additions & 0 deletions tests/sample_data/Package.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
{
"createdByUserId": "usrL2PNC5o3H4lBEi",
"createdTime": "2022-09-12T21:03:48.000Z",
"description": "A new enterprise managed app for the entire grid.",
"enterpriseAccountId": "entUBq2RGdihxl3vU",
"id": "pkggUqk9xHiC4BeeH",
"installCount": 12,
"lastUpdatedByUserId": "usrsOEchC9xuwRgKk",
"lastUpdatedTime": "2022-11-15T01:02:04.400Z",
"latestReleaseId": "pkrsTB7Ic2RhsA4pe",
"name": "New Enterprise Managed App 1",
"sourceApplicationId": "appLkNDICXNqxSDhG",
"tagline": "A new enterprise managed app",
"type": "appTemplate"
}
Loading
Loading