Skip to content

Commit

Permalink
update
Browse files Browse the repository at this point in the history
  • Loading branch information
minenwerfer committed Sep 6, 2023
1 parent f068f01 commit 1cbd86f
Show file tree
Hide file tree
Showing 3 changed files with 18 additions and 174 deletions.
63 changes: 10 additions & 53 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
# Micromodel

Static and runtime dictionary validation (with MongoDB support).
Static and runtime dictionary validation.

## Install

Expand All @@ -23,7 +23,7 @@ Animal = typing.TypedDict('Animal', {
]]
})

# even another TypedDicts can be used!
# another TypedDicts can be nested
Person = typing.TypedDict('Person', {
'name': typing.NotRequired[str | None],
'age': int,
Expand All @@ -34,16 +34,8 @@ m = model(Person, {
'Animal': Animal
})

# hooks can be implemented using monkeypatching
# setting default values also can be achieved this way
old_validate = m.validate
def new_validate(target: Person, options: ValidationOptions = {}):
new = target.copy()
new['name'] = new.get('name', 'unknown')
return old_validate(new, options)

m.validate = new_validate

# the validate method will return the input as it is narrowed as the model type
# or raise if it is invalid
result = m.validate({
'name': 'joao',
'animal': {
Expand All @@ -55,49 +47,14 @@ result = m.validate({
}
})

"""
{
"name": "Joao",
"animal": {
"name": "thor",
"specie": [
"dog"
]
}
}
"""
print(result)

"""
{}
"""
print(m.cast({}))
```

## Usage (with MongoDB)

```python
import os
import typing
from micromodel import model
from pymongo import MongoClient

db = MongoClient(os.getenv('MONGODB_URI')).get_default_database()
# the is_valid method will return the input object narrowed as the model type or
# False otherwise
if valid := m.is_valid(result):
print('dictionary is valid')

Animal = typing.TypedDict('Animal', {
'name': str,
'specie': list[typing.Literal[
'dog',
'cat',
'bird'
]]
})

m = model(Animal, coll=db['animals'])
m.insert_one({
'name': 'thor',
'specie': 'dog'
})
# same as typing.cast(Model[T], {})
print(m.cast({}))
```

## License
Expand Down
72 changes: 8 additions & 64 deletions src/micromodel/micromodel.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,6 @@
import typing
import types
from abc import ABCMeta
from pymongo import ReturnDocument
from pymongo.collection import Collection
from pymongo.results import UpdateResult

T = typing.TypeVar('T')

Expand All @@ -13,72 +10,19 @@
})

class Model(typing.Generic[T]):
coll: Collection

def __init__(self, model_type: typing.Callable[[typing.Any], T], ct: dict[str, typing.Any] = {}, coll: Collection | None = None):
def __init__(self, model_type: typing.Callable[[typing.Any], T], ct: dict[str, typing.Any] = {}):
self.model_type = model_type
self.ct = ct
if coll:
self.coll = coll

def cast(self, target: T | typing.Any): return typing.cast(T, target)
def strict_cast(self, target: T): return target
def validate(self, target: T, options: ValidationOptions = {}): return validate(self.model_type, typing.cast(typing.Any, target), options, self.ct)

def find(self, *args: typing.Any, **kwargs: typing.Any):
result = self.coll.find(*args, **kwargs)
return typing.cast(typing.Generator[T, None, None], result)

def find_one(self, *args: typing.Any, **kwargs: typing.Any):
result = self.coll.find_one(*args, **kwargs)
if not result:
return None
return self.cast(result)

def insert_one(self, what: T, *args: typing.Any, **kwargs: typing.Any):
what = self.validate(what)
result = self.coll.insert_one(typing.cast(typing.Any, what), *args, **kwargs)
return result

def _update(self, value: typing.Any, query_fields: list[str], ret: bool = True):
new = {
k: v
for k, v in value.items()
if k not in [
'_id',
*query_fields
]
}

search = {
'$and': [
{ f: value[f] }
for f in query_fields
if f in value
]
}

if ret:
return self.coll.find_one_and_update(
search,
{ '$set': new },
return_document=ReturnDocument.AFTER,
upsert=True
)
else:
return self.coll.update_one(
search,
{ '$set': new },
upsert=True
)

def update(self, value: typing.Any, query_fields: list[str]):
result = self._update(value, query_fields, ret=True)
return typing.cast(UpdateResult, result)

def upsert(self, value: typing.Any, query_fields: list[str]):
result = self._update(value, query_fields, ret=True)
return typing.cast(T, result)
def is_valid(self, target: T | typing.Any):
try:
return self.validate(target)
except:
return False


def raise_missing_key(k: int | str):
Expand Down Expand Up @@ -157,6 +101,6 @@ def get_hints(model_type: ABCMeta):
hints = typing.get_type_hints(model_type)
return hints

def model(model_type: typing.Callable[[typing.Any], T], ct: dict[str, typing.Any] = {}, coll: Collection | None = None) -> Model[T]:
return Model(model_type, ct, coll)
def model(model_type: typing.Callable[[typing.Any], T], ct: dict[str, typing.Any] = {}) -> Model[T]:
return Model(model_type, ct)

57 changes: 0 additions & 57 deletions tests/mongodb.py

This file was deleted.

0 comments on commit 1cbd86f

Please sign in to comment.