import re
from collections import OrderedDict
from dataclasses import dataclass, field
from pathlib import Path
from textwrap import indent
from typing import Any
import yaml
from pydantic import BaseModel, Field, ValidationInfo, field_validator, model_validator
from botocraft.sync.docstring import DocumentationFormatter
#: The directory containing the data for the botocraft configuration
DATA_DIR = Path(__file__).parent.parent / "data"
#: The directory containing the services for the botocraft configuration
SERVICES_DIR = Path(__file__).parent.parent / "services"
#: The directory containing the documentation for the botocraft configuration
DOCS_DIR = Path(__file__).parent.parent.parent / "doc" / "source"
# ------
# Models
# ------
[docs]class Importable(BaseModel):
"""
A configuration object for a mixin class to add to a model.
"""
#: The name of the mixin class
name: str
#: The botocraft import path
import_path: str
class CodeTransformerDefinition(BaseModel):
"""
Used when generating a property on a model that is a verbatim expression
to return.
"""
#: The name of the attribute on our model to use as
#: the input to the regular expression.
code: str
#: Use this code to transform the attribute value.
return_type: str
[docs]class ModelPropertyDefinition(BaseModel):
"""
The definition of a single property on a :py:class:`ModelDefinition`.
One of ``regex``, ``mapping`` or ``alias`` must be specified.
If :py:attr:`cached` is ``True``, we'll cache the value of this property
using :py:func:`functools.cached_property`.
"""
#: The docstring for this property
docstring: str | None = None
#: If ``True``, make this property be cached
cached: bool = False
#: If specified, use this definition to build the property
#: output from the model fields
transformer: AttributeTransformerDefinition
[docs]class ModelRelationshipDefinition(BaseModel):
"""
Defines a property on a model that is a relationship to another model.
"""
#: If specified, import these modules into the generated file.
imports: list[Importable] = Field(default_factory=list)
#: The docstring for this property
docstring: str | None = None
#: If ``True``, make this property be cached
cached: bool = True
#: If specified, use this definition to build the primary key
#: for the related model from data from this model
#:
#: .. important::
#:
#: If you are using a regex transformer, you must use named groups.
#:
#: Example::
#: r'arn:aws:ecs:(?P<region>[^:]+):(?P<account_id>[^:]+):(?P<resource_type>[^:]+)/(?P<resource_id>[^:]+)' # noqa: E501
transformer: AttributeTransformerDefinition
#: The name of the botocraft primary model to use for this relationship
#:
#: .. important::
#:
#: This needs to be the name of a primary model, not a secondary model.
#: Only primary models have managers.
primary_model_name: str
#: If ``True``, force this to be a many-to-many relationship. Otherwise,
#: try to figure it out from our transformer.
many: bool | None = None
#: If specified, use this as the name of the method on the related model's
#: manager to get the related model instance. This defaults to ``get``
#: if :py:attr:`many` is ``False`` and ``list`` if :py:attr:`many` is ``True``.
method: str | None = None
[docs]class ModelManagerMethodArgDefinition(BaseModel):
#: The name of the argument
name: str
#: The value of the argument. If ``None``, we'll use the value of the
#: attribute on the model with the same name.
value: str | None = None
#: The type of the argument
attr_type: str | None = None
[docs]class ModelManagerMethodKwargDefinition(BaseModel):
#: The name of the argument
name: str
#: The value of the argument. if ``None``, we'll use the value of the
#: attribute on the model with the same name.
value: str | None = None
#: The type of the argument
attr_type: str | None = None
#: The default value for the argument
default: Any | None = None
[docs]class ModelManagerMethodDefinition(BaseModel):
"""
Used to provide methods on a model that call methods on the manager for the
model.
"""
# The name of the method on the manager
manager_method: str
#: If ``True``, make this result from this cached
cached: bool = False
#: The docstring for the method
docstring: str | None = None
#: The arguments for the method. These should refer to attributes : on the
#: model and map them to args on the manager method in the proper : order.
#: The key is the position of the argument in the manager method signature,
#: value is the name of the attribute on the model.
args: dict[int, ModelManagerMethodArgDefinition] = Field(default_factory=dict)
#: Args the users must supply. The key is the position of the argument in
#: the manager method signature, and the value is the name of the argument
#: to add to the model method signature.
user_args: dict[int, ModelManagerMethodArgDefinition] = Field(default_factory=dict)
#: The keyword arguments for the method. These should refer to attributes
#: on the model and map them to kwargs on the manager method with the proper
#: keyword.
keyword_args: list[ModelManagerMethodKwargDefinition] = Field(default_factory=list)
#: Keyword arguments the users can supply
user_keyword_args: list[ModelManagerMethodKwargDefinition] = Field(
default_factory=list
)
#: A list of decorators to wrap the method in
decorators: list[Importable] = Field(default_factory=list)
[docs]class ModelAttributeDefinition(BaseModel):
"""
The definition of a single attribute on a :py:class:`ModelDefinition`.
We use this to override the default configuration of a model attribute
that we generate from the botocore model.
"""
#: If specified, use this as the name of the field in the model.
#: Set the original name of the field as a pydantic field alias.
rename: str | None = None
#: If ``True``, make this field immutable.
readonly: bool = False
#: If ``True``, force the field to be required, even if the underlying
#: botocore field is not required.
required: bool = False
#: If specified, set this as the default value for the field.
default: Any | None = None
#: If specified, use this as the docstring for the field.
docstring: str | None = None
#: If specified, use this as the python type for the field.
python_type: str | None = None
#: If specified, the list of imports to add to the top of the
#: to support the python type.
imports: list[str] = []
#: This is not defined in the YAML file, but is set by
#: :py:meth:`ModelGenerator.fields` during generation.
botocore_shape: Any | None = None
[docs]class ModelDefinition(BaseModel):
"""
The definition of a single model for a :py:class:`ServiceDefinition`.
"""
#: The botocore model name
name: str
#: A list of imports to add to the top of the file to support the model.
imports: list[str] | None = None
#: A list of mixin classes to add to the model
mixins: list[Importable] = Field(default_factory=list)
#: If ``True``, make this model immutable.
readonly: bool = False
#: The plural form of our model name, if different from
#: what the inflect library would generate.
plural: str | None = None
#: The name of the base class for this model. If not specified, we'll
#: use the default base class for the use case.
base_class: str | None = None
#: A different docstring for the model. If not specified, we'll use the
#: docstring from the botocore model.
docstring: str | None = None
#: If defined, use this as the model name instead of what botocore supplies.
#: We use this if some submodels have a ame conflict with a model in
#: another service or an attribute on a model in this service.
alternate_name: str | None = None
#: The primary key for this model. This is the field that will be used
#: to determine whether we need to create a new model instance or update
#: an existing one. This can be the name of a property.
primary_key: str | None = None
#: The ``ARN`` key for this model attribute if any. Some AWS models
#: have an ARN, a name, and an ID. Some have no ARN.
arn_key: str | None = None
#: The ``name`` key for this model attribute if any. Some AWS models
#: have an ARN, a name, and an ID. Some have no name.
name_key: str | None = None
#: A list of field overrides for this model. If a field name is not
#: specified in this list, it will be generated verbatim from the
#: botocore model attribute definition.
fields: dict[str, ModelAttributeDefinition] = Field(default_factory=dict)
#: A list of extra fields for this model. These are fields that are
#: not in the botocore model, but are either are needed at create time
#: or because they turn into something else in the response. This is
#: pretty rarely used.
extra_fields: dict[str, ModelAttributeDefinition] = Field(default_factory=dict)
#: The name of the output shape for the get method for this model. We use
#: this to add to the model any fields that appear in the get method
#: response, but which are not present in the create method response.
output_shape: str | None = None
#: The names of the request shapes for the writable methods for this model.
#: We use these to determine which fields on the model are writable.
input_shapes: list[str] | None = None
#: Computed properties. These are all defined by us in the botocraft
#: config.
properties: dict[str, ModelPropertyDefinition] = Field(default_factory=dict)
#: Relationships to other models. These are all defined by us in the
#: botocraft config.
relations: dict[str, ModelRelationshipDefinition] = Field(default_factory=dict)
#: Methods on the model that call methods on the manager for the model.
manager_methods: dict[str, ModelManagerMethodDefinition] = Field(
default_factory=dict
)
#: If ``True``, this is a bespoke model. This means that we
#: don't have a botocore shape for it and are buiding it out of
#: many AWS API calls. This is used for things like S3 buckets
#: and SQS queues, which have no full shapes in botocore.
bespoke: bool = False
#: If ``True``, this is a secondary model that we want to force to be
#: created during sync. This is used for things like S3 buckets
#: and SQS queues, which have no full shapes in botocore.
force_create: bool = False
[docs] def unalias_field_name(self, field_name: str) -> str:
"""
Get the botocore field name for a field name that may be an alias.
Args:
field_name: the field name or alias to look up
Raises:
KeyError: no such field name or alias
Returns:
The botocore field name
"""
if field_name in self.fields:
return field_name
for botocore_name, field_def in self.fields.items():
if field_def.rename == field_name:
return botocore_name
msg = f'{self.name}: No field or alias named "{field_name}"'
raise KeyError(msg)
# --------
# Managers
# --------
[docs]class MethodArgumentDefinition(BaseModel):
"""
The definition of a single argument on a :py:class:`ManagerMethodDefinition`.
Example:
.. code-block:: yaml
:emphasize-lines: 6,7
Repository:
methods:
create:
boto3_name: create_repository
args:
repositoryName:
required: true
"""
#: If specified, when constructing the boto3 call invocation, set the value
#: of this argument to the value of the model attribute specified
#: here. This is only useful for methods that take a model instance
#: as an argument.
attribute: str | None = None
#: If specified, in the method signature, use this as the argument
#: name. Otherwise, we'll use the name of the boto3 argument.
rename: str | None = None
#: If specified, when constructing the boto3 call invocation, set the value
#: of this argument to the value of this method argument
source_arg: str | None = None
#: If specified, in the boto3 call invocation, set the value
#: of this argument to this specifically. If this is set, we hide
#: the argument from the method signature.
value: Any | None = None
#: If ``True``, take value as the raw value to pass to the boto3 call
#: instead of wrapping it in self.serialize()
raw_value: bool = False
#: If ``True``, force the argument to be in in the method
#: signature. This is only useful for methods that take a model
#: instance as an argument.
explicit: bool = False
#: If ``True``, make this a postional argument. If ``False``, make
#: this a keyword argument.
required: bool = False
#: If specified, set this as the default value for the argument.
default: Any | None = None
#: If ``True``, don't include this argument in the method signature.
#: or in the boto3 call invocation.
hidden: bool = False
#: If specified, use this as the docstring for the argument. Otherwise,
#: we'll use the docstring from the botocore operation.
docstring: str | None = None
#: If supplied, use this as the python type for the argument.
python_type: str | None = None
#: If supplied, this is a list of import strings to add to the top of the
#: of the file to support the python type.
imports: list[str] = Field(default_factory=list)
[docs]class ManagerMethodDefinition(BaseModel):
"""
The definition of a single method on a :py:class:`ManagerDefinition`.
Example:
.. code-block:: yaml
:emphasize-lines: 3,4,5,6,7
Repository:
methods:
create:
boto3_name: create_repository
args:
repositoryName:
required: true
"""
#: The name of the boto3 operation to call in the method body
boto3_name: str
#: A list of argument overrides for this method. If an argument name
#: is not specified as a key in this dict, it will be generated as is
#: appropriate for the operation type
args: dict[str, MethodArgumentDefinition] = {}
#: The attribute name on the response class to use in building our
#: method return object(s). If not specified, we'll use the lowercased
#: model name, pluralizing if necessary.
#:
#: If you want to just return the whole response from the boto3 call,
#: set this to literal ``"None"``.
#:
#: .. important::
#: If you've renamed the model attribute you want to use on the response class
#: using :py:attr:`ModelAttributeDefinition.rename`, put the name of
#: the botocore here, not the value of the ``rename`` attribute.
response_attr: str | None = None
#: If specified, use this as the docstring for the method itself (not
#: the args, kwargs or return type). If not specified, we'll use the
#: docstring from the botocore operation.
docstring: str | None = None
#: If specified, use this as the return type for the method. If not
#: set, we'll use the model name, pluralizing if necessary.
return_type: str | None = None
#: Extra arguments for the method call
extra_args: dict[str, MethodArgumentDefinition] = {}
#: Decorators to wrap the method in
decorators: list[Importable] = []
@property
def explicit_args(self) -> list[str]:
"""
Return a list of positional arguments that should be explicitly
included in the method signature. This is useful for methods
that take a model instance as an argument, but some of the
arguments are not model attributes.
Returns:
A list of explicit positional arguments
"""
if not self.args:
return []
positional_args = []
for arg_name, arg_def in self.args.items():
if arg_def.hidden:
continue
if arg_def.explicit and arg_def.required:
positional_args.append(arg_name)
return positional_args
@property
def explicit_kwargs(self) -> list[str]:
"""
Return a list of keyword arguments that should be explicitly
included in the method signature. This is useful for methods
that take a model instance as an argument, but some of the
arguments are not model attributes.
Returns:
A list of explicit keyword arguments
"""
if not self.args:
return []
positional_args = []
for arg_name, arg_def in self.args.items():
if arg_def.hidden:
continue
if arg_def.explicit and not arg_def.required:
positional_args.append(arg_name)
return positional_args
[docs]@dataclass
class MethodDocstringDefinition:
"""
A class to hold the different parts of the docstring for a method, and
render them into a single sphinx-napoleon style docstring.
"""
#: The main method docstring. This is the description of the method.
method: str | None = None
#: The docstrings for our positional arguments
args: OrderedDict[str, str | None] = field(default_factory=OrderedDict)
#: The docstrings for our keyword arguments
kwargs: OrderedDict[str, str | None] = field(default_factory=OrderedDict)
#: The docstring for our return value
return_value: str | None = None
@property
def is_empty(self) -> bool:
"""
Return ``True`` if the docstring is empty, ``False`` otherwise.
"""
return (
self.method is None
and not self.args
and not self.kwargs
and self.return_value is None
)
[docs] def Args(self, formatter: DocumentationFormatter) -> str:
"""
Return the docstring for the positional arguments.
"""
assert self.args, "No kwargs"
docstring = """
Args:
"""
for arg_name, arg_docstring in self.args.items():
docstring += formatter.format_argument(arg_name, arg_docstring)
docstring += "\n"
return docstring
[docs] def Keyword_Args(self, formatter: DocumentationFormatter) -> str:
"""
Return the docstring for the keyword arguments.
"""
assert self.kwargs, "No args"
docstring = """
Keyword Args:
"""
for arg_name, arg_docstring in self.kwargs.items():
docstring += formatter.format_argument(arg_name, arg_docstring)
docstring += "\n"
return docstring
[docs] def render(self, formatter: DocumentationFormatter) -> str | None:
"""
Return the entire method docstring.
"""
if self.is_empty:
return None
docstring = '''
"""
'''
if self.method:
method_str = formatter.clean(self.method, max_lines=1).strip()
docstring += indent(method_str, " ")
docstring += "\n"
if self.args:
docstring += self.Args(formatter)
if self.kwargs:
docstring += self.Keyword_Args(formatter)
docstring += '\n """'
return docstring
[docs]class ManagerDefinition(BaseModel):
"""
The definition of a single manager on a :py:class:`ServiceDefinition`.
Example:
.. code-block:: yaml
Repository:
methods:
create:
boto3_name: create_repository
args:
repositoryName:
required: true
delete:
boto3_name: delete_repository
args:
repositoryName:
required: true
force:
default: false
"""
#: The name of the model this manager is for
name: str
#: The methods to generate on this manager
methods: dict[str, ManagerMethodDefinition] = {}
#: If ``True``, make this manager use the :py:class:`ReadonlyBoto3ModelManager`
#: superclass
readonly: bool = False
#: Mixin classes to add to the manager
mixins: list[Importable] = []
# --------
# Services
# --------
[docs]class ServiceDefinition(BaseModel):
"""
Our definition of a single AWS service, e.g. ``ecs``, ``ec2``, etc.
"""
#: The global botocraft interface model
interface: "BotocraftInterface"
#: The name of the botocore service
name: str
#: The models to generate for this service. These are specifically models
#: that have managers.
primary_models: dict[str, ModelDefinition] = {}
#: These are models that are used as attributes on other models, or as
#: response or request classes. These models don't have managers, but
#: we sometimes define them to override the default botocore model
#: configuration, making them readonly or to rename them, for instance.
secondary_models: dict[str, ModelDefinition] = {}
#: The managers to generate for this service
managers: dict[str, ManagerDefinition] = {}
@property
def models(self) -> dict[str, ModelDefinition]:
"""
Return all of the models we've specifically named in the definition for
this service, both primary and secondary.
.. important::
This does not include models that we generate on the fly, like
dependent models for primary models, or models for the request and
response classes for methods on managers.
"""
return {**self.primary_models, **self.secondary_models}
[docs] @classmethod
def load(cls, name: str, interface: "BotocraftInterface") -> "ServiceDefinition":
"""
Load a service definition from its YAML files.
Args:
name: The name of the AWS service to load
interface: The global botocraft interface model
Returns:
The loaded service definition
"""
managers: dict[str, ManagerDefinition] | None = None
model_path = DATA_DIR / name / "models.yml"
manager_path = DATA_DIR / name / "managers.yml"
with model_path.open(encoding="utf-8") as f:
models = yaml.safe_load(f)
if manager_path.exists():
with manager_path.open(encoding="utf-8") as f:
managers = {
name: ManagerDefinition(name=name, **defn)
for name, defn in yaml.safe_load(f).items()
}
return cls(
name=name,
primary_models={
name: ModelDefinition(name=name, **defn)
for name, defn in models.get("primary", {}).items()
},
secondary_models={
name: ModelDefinition(name=name, **defn)
for name, defn in models.get("secondary", {}).items()
},
managers=managers,
interface=interface,
)
[docs] def resolve_model_name(self, model_name: str) -> str:
"""
Resolve a model name to its botocore shape name. We need to
do this because we can define :py:attr:`ModelDefinition.alternate_name`
to change the botocraft name for a model.
Args:
model_name: The botocraft model name
Raises:
KeyError: If the model name is not defined in this service
Returns:
The botocore shape name
"""
known_models = {**self.primary_models, **self.secondary_models}
if model_name in known_models:
return model_name
for name, model in known_models.items():
if model.alternate_name == model_name:
return name
msg = f'Unknown model name "{model_name}"'
raise KeyError(msg)
@property
def safe_service_name(self) -> str:
"""
Return a safe version of the service name. This is used to
generate the import path for the service.
Returns:
The safe service name
"""
return self.name.replace("-", "_")
# ------
# Global
# ------
class BotocraftInterface(BaseModel):
#: The services to generate
services: dict[str, ServiceDefinition] = {}
#: The model name to import path mappings
models: dict[str, str] = {
# Tag is defined in so many services that we manually
# defined it in the botocraft.models.tagging module and
# define it here so it doesn't get generated.
"Tag": "botocraft.services.common",
"Filter": "botocraft.services.common",
}
def load(self) -> None:
"""
Load all known service definitions from their YAML files.
Returns:
The loaded interface.
"""
for path in DATA_DIR.iterdir():
service = path.name
if path.is_dir():
self.services[service] = ServiceDefinition.load(service, interface=self)
for model_name, model in self.services[service].primary_models.items():
# Register the model name to import path mapping so that we can
# have them available when we're building our inter-model
# relationships.
_model_name = model.alternate_name or model_name
if _model_name in self.models:
msg = f'Model {_model_name} already defined "'
f'in "{self.models[_model_name]}"'
raise ValueError(msg)
self.models[_model_name] = (
f"botocraft.services.{self.services[service].safe_service_name}"
)
for manager in self.services[service].managers:
# Register the manager name to import path mapping so that we can
# have them available when we're building our inter-model
# relationships.
_manager = manager
if manager in self.services[service].primary_models:
# We already registered this model during :py:meth:`load`
model = self.services[service].primary_models[manager]
if model.alternate_name:
_manager = model.alternate_name
_manager = f"{_manager}Manager"
self.models[_manager] = (
f"botocraft.services.{self.services[service].safe_service_name}"
)
def add_model(self, name: str, service: str) -> None:
"""
Save a model name to import path mapping.
Args:
name: the model name
service: the AWS service name
"""
if name in self.models:
if (
self.models[name]
== f"botocraft.services.{self.services[service].safe_service_name}"
):
# We already registered this model during :py:meth:`load`
return
msg = f'Model {name} already defined in "{self.models[name]}"'
raise ValueError(msg)
self.models[name] = (
f"botocraft.services.{self.services[service].safe_service_name}"
)
def populate_init_py(self):
"""
Populate the ``__init__.py`` file in the ``botocraft.models``
package with import statements for all of our models.
"""
init_path = SERVICES_DIR / "__init__.py"
with init_path.open("w", encoding="utf-8") as f:
for service in self.services.values():
f.write(
f"from .{service.safe_service_name} import * # noqa: F401,F403\n"
)
def populate_services_toc(self):
"""
Populate the ``_services_index.rst`` file in the ``doc/source``
folder with the ``.. toctree::`` directive for all of our services.
"""
index_path = DOCS_DIR / "_services_index.rst"
code = """
.. toctree::
:caption: AWS Services
:hidden:
"""
for service in sorted(self.services):
code += f" api/services/{self.services[service].safe_service_name}\n"
with index_path.open("w", encoding="utf-8") as f:
f.write(code)
def populate_services_list(self):
"""
Populate the ``_services_list.rst`` file in the ``doc/source``
folder with links to all of our services.
"""
index_path = DOCS_DIR / "overview" / "_services_list.rst"
code = ""
for service in sorted(self.services):
code += (
f"- :doc:`/api/services/{self.services[service].safe_service_name}`\n"
)
with index_path.open("w", encoding="utf-8") as f:
f.write(code)
def update_init_py(self, service: str) -> None:
"""
Update the ``__init__.py`` file in the ``botocraft.models``
package with import statements for the given service.
Args:
service: The name of the service to update the ``__init__.py``
file for.
"""
init_path = SERVICES_DIR / "__init__.py"
import_line = (
f"from .{self.services[service].safe_service_name} import * "
"# noqa: F401,F403"
)
with init_path.open(encoding="utf-8") as f:
contents = f.read()
if import_line not in contents:
with init_path.open("a", encoding="utf-8") as f:
f.write(f"{import_line}\n")
def update_services_toc(self, service: str) -> None:
"""
Update the services toctree with the given service if it is not present.
Args:
service: The name of the service to add to the services toctree.
"""
path = DOCS_DIR / "_services_index.rst"
line = f" api/services/{self.services[service].safe_service_name}"
if not path.exists():
self.populate_services_toc()
return
with path.open(encoding="utf-8") as f:
contents = f.read()
if line not in contents:
with path.open("a", encoding="utf-8") as f:
f.write(f"{line}\n")
def update_services_list(self, service: str) -> None:
"""
Update the services overview list with the given service if it is not
present.
Args:
service: The name of the service to add to the overview list.
"""
path = DOCS_DIR / "overview" / "_services_list.rst"
line = f"- :doc:`/api/services/{self.services[service].safe_service_name}`"
if not path.exists():
self.populate_services_list()
return
with path.open(encoding="utf-8") as f:
contents = f.read()
if line not in contents:
with path.open("a", encoding="utf-8") as f:
f.write(f"{line}\n")
def purge_docs(self) -> None:
"""
Purge all the previously generated documentation.
"""
path = DOCS_DIR / "api" / "services"
for service in self.services.values():
service_path = path / f"{service.safe_service_name}.rst"
if service_path.exists():
service_path.unlink()
path = DOCS_DIR / "_services_index.rst"
if path.exists():
path.unlink()
path = DOCS_DIR / "overview" / "_services_list.rst"
if path.exists():
path.unlink()
def generate(self, service: str | None = None) -> None:
"""
Generate all our service interfaces.
Keyword Args:
service: If specified, only generate the interface for the given
service. Otherwise, generate all interfaces.
"""
from .service import ServiceGenerator
if service:
assert service in self.services, (
f'No service definition for AWS Service "{service}"'
)
print(f"Generating {service} service interface")
generator = ServiceGenerator(self.services[service])
generator.generate()
self.update_init_py(service)
self.update_services_toc(service)
self.update_services_list(service)
else:
# First purge all the previously generated documentation
self.purge_docs()
for service_name, _service in self.services.items():
print(f"Generating {service_name} service interface")
generator = ServiceGenerator(_service)
generator.generate()
self.populate_init_py()
self.populate_services_toc()
self.populate_services_list()