| Name | noverde-serpens JSON |
| Version |
2.7.0
JSON |
| download |
| home_page | None |
| Summary | A set of Python utilities, recipes and snippets |
| upload_time | 2024-09-23 21:15:14 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.8 |
| license | MIT |
| keywords |
utilities
aws
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# serpens
A set of Python utilities, recipes and snippets.
- [SQS Utilities](#sqs-utilities)
- [API Utilities](#api-utilities)
- [Schema](#schema)
- [CSV Utils](#csv-utils)
- [DynamoDB Documents](#dynamodb-documents)
## SQS Utilities
- This utility is a decorator that iterate by each sqs record.
- For each sqs record will be inserted a *record object (from type sqs.Record)* as argument that will process the sqs messages.
```python
from serpens import sqs
@sqs.handler
def message_processor(record: sqs.Record):
# code to process each sqs message
print(record.body)
```
### Record
- The client function that will process the sqs messages will receive an instance of *sqs.Record* dataclass. This class has the follow structure:
```python
class Record:
data: Dict[Any, Any]
body: Union[dict, str]
message_attributes: Dict[Any, Any]
queue_name: str
sent_datetime: datetime
```
##### Record attributes
- **data**: Contain all data from SQS message. This attribute is assigned in each iteration in SQS message.
- **body**: Return ```data["body"]``` converted to ```dict``` or ```str```.
- **message_attributes**: Return the ```data["messageAttributes"]``` converted to ```dict```.
- **queue_name**: Return the queue name extracted from ```data["eventSourceARN"]```.
- **sent_datetime**: Return the ```data["attributes"]["SentTimestamp"]``` converted to ```datetime```.
## API Utilities
- This utility is a wrapper to simplify working with lambda handlers. The decorator ```api.handler``` will decorate a function that will process a lambda and this function will receive a ```request``` argument (from type api.Request).
```python
from serpens import api
@api.handler
def lambda_handler(request: api.Request):
# Code to process the lambda
print(request.body)
```
#### *Request class*
- The function that will process the lambda will receive an instance of *sqs.Request* dataclass. This class has the follow structure:
```python
from serpens.api import AttrDict
class Request:
authorizer: AttrDict
body: Union[str, dict]
path: AttrDict
query: AttrDict
headers: AttrDict
identity: AttrDict
```
- *Note*: the objects from type ```AttrDict``` are objects built by a dict where the dict's key is an attribute of object. For example:
```python
from serpens.api import AttrDict
obj = AttrDict({"foo": "bar"})
obj.foo # bar
```
## Schema
- The Schema is a base class for create new classes with follow features:
> - Static type check
> - Method to convert an object to dict
> - Method to create an object from json
> - Method to create an object from dict
> - Method to dump an object to string
##### Create a schema
```python
from serpens.schema import Schema
from dataclasses import dataclass
@dataclass
class PersonSchema(Schema):
name: str
age: int
```
##### Create a schema object
```python
person = PersonSchema('Mike', 30)
print(person.name)
print(person.age)
```
##### Create a schema object from a dict.
```python
person_obj = PersonSchema.load({'name': 'Mike', 'age': 18})
print(person_obj.name)
print(person_obj.age)
```
##### Create a schema object from a json string.
```python
import json
data = json.dumps({'name': 'mike', 'age': 20})
person_obj = PersonSchema.loads(data)
print(person_obj.name)
print(person_obj.age)
```
##### Convert a schema object to dict.
```python
p1 = PersonSchema('Mike', 30)
person_dct = PersonSchema.dump(p1)
print(person_dct['name'])
print(person_dct['age'])
```
##### Convert a schema object to json string.
```python
p1 = PersonSchema('Mike', 30)
person_str = PersonSchema.dumps(p1)
print(person_str)
```
## CSV Utils
- Utility for read and write csv. This utility is useful for read csv with BOM or read csv encoded as ISO-8859.
##### Read CSV
```python
from serpens import csvutils as csv
dict_reader = csv.open_csv_reader('fruits_iso8859.csv')
for record in dict_reader:
print(record)
```
##### Write CSV
```python
from serpens import csvutils as csv
writer = csv.open_csv_writer('out.csv')
writer.writerow(["id", "name"])
writer.writerow(["1", "Açaí"])
del writer
```
## Database
This utilities are useful for working with database.
##### Migrate databases
- This migrations use yoyo-migration.
```python
from serpens import database
database_url = "postgres://user:password@host/db"
path = "/path/to/migrations" # yoyo migrations
database.migrate(database_url, path)
```
##### Create a Pony Database instance
"*The Database object manages database connections using a connection pool.*"
```python
from serpens import database
database_url = "postgres://user:password@host/db"
db = database.setup(database_url)
print(db.provider_name)
```
## DynamoDB Documents
Serpens provides a base class (called *BaseDocument*) for working with tables from DynamoDB.
##### Create a document mapping a DynamoDB table
```python
from serpens.document import BaseDocument
from dataclasses import dataclass
@dataclass
class PersonDocument(BaseDocument):
_table_name_ = 'person'
id: str
name: str
```
##### Save data in DynamoDB table
```python
person = PersonDocument(id="1", name="Ana")
person.save()
```
##### Get data from key
- Obs: If the search doesn't find an item by its key, the return is ```None```
```python
person = PersonDocument.get_by_key({"id": "1"})
person.id # 1
person.name # Ana
```
##### Get table
```python
person_table = PersonDocument.get_table()
person_table # dynamodb.Table(name='person')
```
Raw data
{
"_id": null,
"home_page": null,
"name": "noverde-serpens",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "utilities, aws",
"author": null,
"author_email": "Noverde Developers <dev@noverde.com.br>",
"download_url": "https://files.pythonhosted.org/packages/da/3b/12f3df0b02f9ddaa6789afc1a82edd06deeb5edd89344a8619fa8a82e8ad/noverde_serpens-2.7.0.tar.gz",
"platform": null,
"description": "# serpens\n\nA set of Python utilities, recipes and snippets.\n\n- [SQS Utilities](#sqs-utilities)\n- [API Utilities](#api-utilities)\n- [Schema](#schema)\n- [CSV Utils](#csv-utils)\n- [DynamoDB Documents](#dynamodb-documents)\n\n## SQS Utilities\n\n- This utility is a decorator that iterate by each sqs record.\n\n- For each sqs record will be inserted a *record object (from type sqs.Record)* as argument that will process the sqs messages.\n\n\n```python\nfrom serpens import sqs\n\n@sqs.handler\ndef message_processor(record: sqs.Record):\n # code to process each sqs message\n print(record.body)\n```\n\n### Record\n\n- The client function that will process the sqs messages will receive an instance of *sqs.Record* dataclass. This class has the follow structure:\n\n```python\nclass Record:\n data: Dict[Any, Any]\n body: Union[dict, str]\n message_attributes: Dict[Any, Any]\n queue_name: str\n sent_datetime: datetime\n```\n##### Record attributes\n- **data**: Contain all data from SQS message. This attribute is assigned in each iteration in SQS message.\n- **body**: Return ```data[\"body\"]``` converted to ```dict``` or ```str```.\n- **message_attributes**: Return the ```data[\"messageAttributes\"]``` converted to ```dict```.\n- **queue_name**: Return the queue name extracted from ```data[\"eventSourceARN\"]```.\n- **sent_datetime**: Return the ```data[\"attributes\"][\"SentTimestamp\"]``` converted to ```datetime```.\n\n## API Utilities\n\n- This utility is a wrapper to simplify working with lambda handlers. The decorator ```api.handler``` will decorate a function that will process a lambda and this function will receive a ```request``` argument (from type api.Request).\n\n\n```python\nfrom serpens import api\n\n@api.handler\ndef lambda_handler(request: api.Request):\n # Code to process the lambda\n print(request.body)\n```\n\n#### *Request class*\n\n- The function that will process the lambda will receive an instance of *sqs.Request* dataclass. This class has the follow structure:\n\n```python\nfrom serpens.api import AttrDict\n\nclass Request:\n authorizer: AttrDict\n body: Union[str, dict]\n path: AttrDict\n query: AttrDict\n headers: AttrDict\n identity: AttrDict\n```\n\n- *Note*: the objects from type ```AttrDict``` are objects built by a dict where the dict's key is an attribute of object. For example:\n\n\n```python\nfrom serpens.api import AttrDict\n\nobj = AttrDict({\"foo\": \"bar\"})\nobj.foo # bar\n```\n\n## Schema\n- The Schema is a base class for create new classes with follow features:\n> - Static type check\n> - Method to convert an object to dict\n> - Method to create an object from json\n> - Method to create an object from dict\n> - Method to dump an object to string\n\n##### Create a schema\n\n```python\nfrom serpens.schema import Schema\nfrom dataclasses import dataclass\n\n@dataclass\nclass PersonSchema(Schema):\n name: str\n age: int\n```\n##### Create a schema object\n\n```python\nperson = PersonSchema('Mike', 30)\n\nprint(person.name)\nprint(person.age)\n```\n\n##### Create a schema object from a dict.\n\n```python\nperson_obj = PersonSchema.load({'name': 'Mike', 'age': 18})\n\nprint(person_obj.name)\nprint(person_obj.age)\n```\n\n##### Create a schema object from a json string.\n\n```python\nimport json\ndata = json.dumps({'name': 'mike', 'age': 20})\nperson_obj = PersonSchema.loads(data)\n\nprint(person_obj.name)\nprint(person_obj.age)\n```\n\n##### Convert a schema object to dict.\n\n```python\np1 = PersonSchema('Mike', 30)\nperson_dct = PersonSchema.dump(p1)\n\nprint(person_dct['name'])\nprint(person_dct['age'])\n```\n\n##### Convert a schema object to json string.\n\n```python\np1 = PersonSchema('Mike', 30)\nperson_str = PersonSchema.dumps(p1)\n\nprint(person_str)\n```\n\n## CSV Utils\n\n- Utility for read and write csv. This utility is useful for read csv with BOM or read csv encoded as ISO-8859. \n\n##### Read CSV\n\n```python\nfrom serpens import csvutils as csv\n\ndict_reader = csv.open_csv_reader('fruits_iso8859.csv')\n\nfor record in dict_reader:\n print(record)\n```\n\n##### Write CSV\n\n```python\nfrom serpens import csvutils as csv\n\nwriter = csv.open_csv_writer('out.csv')\nwriter.writerow([\"id\", \"name\"])\nwriter.writerow([\"1\", \"A\u00e7a\u00ed\"])\n\ndel writer\n```\n\n## Database\n\nThis utilities are useful for working with database.\n\n##### Migrate databases\n\n- This migrations use yoyo-migration.\n\n```python\nfrom serpens import database\n\ndatabase_url = \"postgres://user:password@host/db\"\npath = \"/path/to/migrations\" # yoyo migrations\n\ndatabase.migrate(database_url, path)\n```\n\n##### Create a Pony Database instance\n\n\"*The Database object manages database connections using a connection pool.*\"\n\n```python\nfrom serpens import database\n\ndatabase_url = \"postgres://user:password@host/db\"\ndb = database.setup(database_url)\nprint(db.provider_name)\n```\n\n## DynamoDB Documents\n\nSerpens provides a base class (called *BaseDocument*) for working with tables from DynamoDB. \n\n##### Create a document mapping a DynamoDB table\n\n```python\nfrom serpens.document import BaseDocument\nfrom dataclasses import dataclass\n\n@dataclass\nclass PersonDocument(BaseDocument):\n _table_name_ = 'person'\n id: str\n name: str\n```\n\n##### Save data in DynamoDB table\n\n```python\nperson = PersonDocument(id=\"1\", name=\"Ana\")\nperson.save()\n```\n\n##### Get data from key\n\n- Obs: If the search doesn't find an item by its key, the return is ```None```\n\n```python\nperson = PersonDocument.get_by_key({\"id\": \"1\"})\n\nperson.id # 1\nperson.name # Ana\n```\n\n##### Get table\n\n```python\nperson_table = PersonDocument.get_table()\nperson_table # dynamodb.Table(name='person')\n```\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A set of Python utilities, recipes and snippets",
"version": "2.7.0",
"project_urls": null,
"split_keywords": [
"utilities",
" aws"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b778a6550f2c64be1eba4f55f2e6043f9f1264f780291b54b8cbce984a61248c",
"md5": "8e09ea20cee9939d591b01c0db607d21",
"sha256": "f7e1f4408206bcca0e823b8c0d9bd9e1daec7dc29bbb9477bccf0b45b1f05afd"
},
"downloads": -1,
"filename": "noverde_serpens-2.7.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "8e09ea20cee9939d591b01c0db607d21",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 22970,
"upload_time": "2024-09-23T21:15:12",
"upload_time_iso_8601": "2024-09-23T21:15:12.538663Z",
"url": "https://files.pythonhosted.org/packages/b7/78/a6550f2c64be1eba4f55f2e6043f9f1264f780291b54b8cbce984a61248c/noverde_serpens-2.7.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "da3b12f3df0b02f9ddaa6789afc1a82edd06deeb5edd89344a8619fa8a82e8ad",
"md5": "6d7a783fd7c2c8819276d0949199a491",
"sha256": "abc370500d81054f945fc91a692298ffcd4680e1deffde9bb7de8c5f50fb2ba7"
},
"downloads": -1,
"filename": "noverde_serpens-2.7.0.tar.gz",
"has_sig": false,
"md5_digest": "6d7a783fd7c2c8819276d0949199a491",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 32841,
"upload_time": "2024-09-23T21:15:14",
"upload_time_iso_8601": "2024-09-23T21:15:14.048163Z",
"url": "https://files.pythonhosted.org/packages/da/3b/12f3df0b02f9ddaa6789afc1a82edd06deeb5edd89344a8619fa8a82e8ad/noverde_serpens-2.7.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-23 21:15:14",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "noverde-serpens"
}
