{
  "markdown": "# Prismarine - DynamoDB ORM\n\nPrismarine is a Pythonic ORM for DynamoDB, designed to simplify interactions with DynamoDB by providing a structured and Python-friendly interface. It leverages Python's type hinting and decorators to define models, which are then used to generate client code for database operations.\n\nKey features include:\n- **Model Definition**: Models are defined using Python's `TypedDict` (default) or, optionally, `pydantic.BaseModel` classes and are decorated with the `Cluster.model` decorator to specify primary and sort keys.\n- **Automatic Client Generation**: The `prismarine_client.py` file is auto-generated, containing classes and methods for interacting with DynamoDB tables based on the defined models.\n- **Easy Integration**: The generated client code integrates seamlessly with existing Python applications, providing methods for common database operations.\n\nPrismarine aims to streamline the development process by reducing boilerplate code and ensuring that database interactions are type-safe and maintainable.\n\nPrismarine works best with [EasySAM](https://github.com/scartill/easysam).\n\n## Installation\n\n```bash\npip install prismarine\n```\n\n## Quick Overview\n\n### Expected Directory Structure:\n\n```\n<base-path>/\n  <package-name>/\n    - models.py\n    - db.py\n    - prismarine_client.py // Auto-generated\n```\n\nModels are defined in the `models.py` file. Each model is a `TypedDict`, decorated with the `Cluster.model` decorator. You can also opt into Pydantic models (see [Using Pydantic Models](#using-pydantic-models)).\n\nThe `Cluster` class is used to group extension models together. It also sets a prefix for the table names.\n\n```python\nfrom typing import TypedDict, NotRequired\nfrom prismarine import Cluster\n\nc = Cluster('TapgameExample')\n\n@c.model(PK='Foo', SK='Bar')\nclass Team(TypedDict):\n    Foo: str\n    Bar: str\n    Baz: NotRequired[str]\n```\n\nIf we place this code in `<base-path>/<package-name>/models.py` and the following command is run, it will generate a `prismarine_client.py` file in the same directory:\n\n```bash\nprismarine generate-client --base <base-path> <package-name>\n```\n\nThe `prismarine_client.py` file will contain the following code:\n\n```python\nclass TeamModel(Model):\n    table_name = 'TapgameExampleTeam'\n    PK = 'Foo'\n    SK = 'Bar'\n\n    class UpdateDTO(TypedDict, total=False):\n        Foo: str\n        Bar: str\n        Baz: NotRequired[str]\n\n    @staticmethod\n    def list(*, foo: str) -> List[Team]:\n        ...\n\n    @staticmethod\n    def get(*, bar: str, foo: str, default: Team | EllipsisType = ...) -> Team:\n        ...\n\n    @staticmethod\n    def put(team: Team) -> Team:\n        ...\n\n    @staticmethod\n    def update(\n        team: UpdateDTO, *, foo: str, bar: str, default: Team | EllipsisType = ...\n    ) -> Team:\n        ...\n\n    @staticmethod\n    def save(updated: Team, *, original: Team | None = None) -> Team:\n        ...\n\n    @staticmethod\n    def delete(*, bar: str, foo: str):\n        ...\n\n    @staticmethod\n    def scan() -> List[Team]:\n        ...\n```\n\nAs you can see, the `TeamModel` class has static methods for all the CRUD operations. The `UpdateDTO` class is similar to the `Team` class, but all fields are optional.\n\n### Creating a `db.py` File\n\nNow, let's create a `db.py` file in the same directory:\n\n```python\nimport example.prismarine_client as pc\n\nclass TeamModel(pc.TeamModel):\n    pass\n```\n\nAlthough you can import and use `prismarine_client.py` directly, it is recommended to create a `db.py` file that imports the generated client and extends it with your own methods.\n\nYou can now use the `TeamModel` class in your code:\n\n```python\nfrom sam.common.example.db import TeamModel\nfrom sam.common.prismarine import DbNotFound\n\n# Create a new team\nnew_team = TeamModel.put({'Foo': 'foo', 'Bar': 'bar', 'Baz': 'baz'})\n\n# List teams by a primary key\nteams_by_foo = TeamModel.list(foo='foo')\n\n# Get a team\ntry:\n    team = TeamModel.get(foo='foo', bar='bar')\nexcept DbNotFound:\n    print('Team not found')\n\n# Update a team\nupdated_team = TeamModel.update(\n    {'Baz': 'new_baz'},\n    foo='foo',\n    bar='bar'\n)\n\n# List all teams\nall_teams = TeamModel.scan()\n\n# Delete a team\nTeamModel.delete(foo='foo', bar='bar')\n```\n\nYou may notice that Prismarine mostly requires named arguments. This ensures that changes to field names do not cause silent code failures. For example, if the Sort Key name is changed, all usages of `get` and `update` methods will break and be highlighted by the IDE and linter. This approach also makes the code more readable.\n\n### Using Pydantic Models\n\nPrismarine can optionally generate clients that work with `pydantic.BaseModel` schemas rather than `TypedDict`.\n\n1. Install the optional dependency:\n\n```bash\npip install \"prismarine[pydantic]\"\n```\n\n2. Define your models as `BaseModel` subclasses in `models.py`.\n3. Run the generator with the Pydantic model library enabled:\n\n```bash\nprismarine generate-client --model-library pydantic --base <base-path> <package-name>\n```\n\nWith this flag disabled (the default `typed-dict` mode), Prismarine behaves exactly as before. The Pydantic mode keeps the same API surface but returns/accepts BaseModel instances and automatically converts data during CRUD operations.\n\n## Advanced Usage\n\n### `model` Decorator\n\nThe `Cluster.model` decorator accepts several arguments to customize the model:\n\n- **`PK`** (required): The name of the partition key attribute\n- **`SK`** (optional): The name of the sort key attribute\n- **`table`** (optional): Sets a full custom table name (without prefix)\n- **`name`** (optional): Sets a custom model name (used with prefix)\n- **`trigger`** (optional): Configures a DynamoDB stream trigger for the table (when using with EasySAM)\n- **`ttl`** (optional): Configures a DynamoDB Time To Live (TTL) attribute for the table (when using with EasySAM)\n\nFor example, if the `Cluster` has a prefix `TapgameExample`, by default the `Team` model will have the table name `TapgameExampleTeam`. If we set `name='Custom'`, the table name will be `TapgameExampleCustom`. And if we set `table='CustomTable'`, the table name will simply be `CustomTable`, without the prefix.\n\n#### DynamoDB Stream Triggers\n\nWhen using Prismarine with [EasySAM](https://github.com/scartill/easysam), you can configure DynamoDB stream triggers directly on your models using the `trigger` parameter. This allows a Lambda function to be automatically invoked whenever items in the table are inserted, modified, or removed.\n\n**Simple trigger (string format):**\n\n```python\n@c.model(PK='Foo', SK='Bar', trigger='itemlogger')\nclass Item(TypedDict):\n    Foo: str\n    Bar: str\n```\n\n**Advanced form** (with options):\n\n```python\n@c.model(\n    PK='Foo',\n    SK='Bar',\n    trigger={\n        'function': 'my-lambda',\n        'viewtype': 'new-and-old',  # Optional: keys-only, new, old, new-and-old (default: new-and-old)\n        'batchsize': 10,             # Optional: number of records per batch\n        'batchwindow': 5,            # Optional: seconds to wait for batch\n        'startingposition': 'latest' # Optional: trim-horizon, latest (default: latest)\n    }\n)\nclass Item(TypedDict):\n    Foo: str\n    Bar: str\n```\n\nThe trigger configuration options:\n- **function**: The name of the Lambda function to trigger\n- **viewtype**: What data to include in the stream record (default: `new-and-old`)\n  - `keys-only`: Only the keys of the modified item\n  - `new`: Only the new item image\n  - `old`: Only the old item image\n  - `new-and-old`: Both old and new item images\n- **batchsize**: Number of records to process per batch (improves throughput)\n- **batchwindow**: Maximum number of seconds to wait for a batch (reduces latency)\n- **startingposition**: Where to start reading the stream (default: `latest`)\n  - `trim-horizon`: Start from the oldest record available\n  - `latest`: Start from the most recent record\n\nWhen EasySAM generates the CloudFormation template, it will automatically:\n- Enable DynamoDB Streams on the table\n- Create an EventSourceMapping to connect the stream to your Lambda function\n- Configure the appropriate IAM permissions for stream access\n\nThe trigger Lambda function will receive DynamoDB stream events with information about inserted, modified, or removed items.\n\n#### DynamoDB Time To Live (TTL)\n\nWhen using Prismarine with [EasySAM](https://github.com/scartill/easysam), you can configure DynamoDB Time To Live (TTL) directly on your models using the `ttl` parameter. This allows DynamoDB to automatically delete items after a specified expiration time.\n\n**Example:**\n\n```python\nfrom typing import TypedDict, NotRequired\nfrom prismarine.runtime import Cluster\n\nc = Cluster('PrismaTTL')\n\n@c.model(PK='Foo', SK='Bar', ttl='ExpireAt')\nclass Item(TypedDict):\n    Foo: str\n    Bar: str\n    Baz: NotRequired[str]\n    ExpireAt: int  # Unix timestamp (seconds since epoch)\n```\n\nThe `ttl` parameter specifies the attribute name that will store the expiration timestamp. When you create or update an item, set this attribute to a Unix timestamp (number of seconds since epoch). DynamoDB will automatically delete items within 48 hours after the TTL timestamp has passed.\n\n**Benefits:**\n- **Automatic Cleanup**: Items are automatically deleted without additional code\n- **Cost Effective**: TTL deletion is free and doesn't consume write capacity units\n- **Declarative**: Define TTL directly in your model configuration\n\nWhen EasySAM generates the CloudFormation template, it will automatically:\n- Enable TTL on the DynamoDB table\n- Configure the `TimeToLiveSpecification` with the specified attribute name\n\n### `index` Decorator\n\n`index` decorators must be used *before* the `model` decorator.\n\nThe `Cluster.index` decorator is used to define a secondary index. It accepts `PK`, `SK`, and `index` arguments.\n\n```python\n@c.index(index='by-bar', PK='Bar', SK='Foo')\n@c.model(PK='Foo', SK='Bar')\nclass Team(TypedDict):\n    Foo: str\n    Bar: str\n    Baz: NotRequired[str]\n```\n\nThis will add a subclass `ByBar` to the `TeamModel` class:\n\n```python\nclass TeamModel(Model):\n    ...\n\n    class ByBar:\n        PK = 'Bar'\n        SK = 'Foo'\n\n        @staticmethod\n        def list(\n            *,\n            bar: str,\n            limit: int | None = None,\n            direction: Literal['ASC', 'DESC'] = 'ASC'\n        ) -> List[Team]:\n            ...\n\n        @staticmethod\n        def get(*, bar: str, foo: str) -> Team:\n            ...\n```\n\n### `export` Decorator\n\nThe `Cluster.export` decorator is used to define a class that is not a model, but is exported from the cluster. It accepts a class as an argument. It is required to used on all classes that serve as types for model elements.\n\n```python\n@c.export\nclass Team(TypedDict):\n    Foo: str\n    Bar: str\n```\n\n## Other Commands\n\n### `version`\n\nPrints the version of Prismarine.\n\n```bash\nprismarine version\n```\n",
  "bytes": 10821,
  "sha": "ace0dd1f84059a59850c891a217f7ef6179f653ec84066a6ecb6b8e205b94a7e",
  "repo_slug": "scartill/prismarine",
  "fonte": "repo",
  "truncated": false,
  "api": "https://agentalog.com/api/listings/okf_scartill_prismarine_openwiki_index_md_d3887cce/readme"
}