A set of Python utilities, recipes and snippets
Project description
serpens
A set of Python utilities, recipes and snippets.
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.
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:
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 todict
orstr
. - message_attributes: Return the
data["messageAttributes"]
converted todict
. - queue_name: Return the queue name extracted from
data["eventSourceARN"]
. - sent_datetime: Return the
data["attributes"]["SentTimestamp"]
converted todatetime
.
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 arequest
argument (from type api.Request).
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:
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:
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
from serpens.schema import Schema
from dataclasses import dataclass
@dataclass
class PersonSchema(Schema):
name: str
age: int
Create a schema object
person = PersonSchema('Mike', 30)
print(person.name)
print(person.age)
Create a schema object from a dict.
person_obj = PersonSchema.load({'name': 'Mike', 'age': 18})
print(person_obj.name)
print(person_obj.age)
Create a schema object from a json string.
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.
p1 = PersonSchema('Mike', 30)
person_dct = PersonSchema.dump(p1)
print(person_dct['name'])
print(person_dct['age'])
Convert a schema object to json string.
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
from serpens import csvutils as csv
dict_reader = csv.open_csv_reader('fruits_iso8859.csv')
for record in dict_reader:
print(record)
Write CSV
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.
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."
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
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
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
person = PersonDocument.get_by_key({"id": "1"})
person.id # 1
person.name # Ana
Get table
person_table = PersonDocument.get_table()
person_table # dynamodb.Table(name='person')
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
noverde-serpens-2.2.0.tar.gz
(27.5 kB
view hashes)
Built Distribution
Close
Hashes for noverde_serpens-2.2.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 9c232aa25939d461d8f7f93b12cfe3f1141c50050b947fc8c16ea7a8297cf4e3 |
|
MD5 | e05bf92c63955be342898de87fd9fb95 |
|
BLAKE2b-256 | a18ae4cb44e163a86eb622618ef104ec4e0614b5454073c3b899f36d17387d27 |