ODM - Object Document Mapper (via mongoengine)

Imports



In [ ]:

    
import mongoengine
from mongoengine import *

Register MongoEngine Connection



In [ ]:

    
connection_info = {
    'host': 'localhost',
    'port': 32768,
    # 'username': user,
    # 'password': password,
    # 'authentication_source': 'admin',
    # 'authentication_mechanism': 'SCRAM-SHA-1',
    # 'ssl': use_ssl,
    # 'ssl_cert_reqs': ssl.CERT_NONE,
}
mongoengine.register_connection(alias='core', name='mongo_meetup', **connection_info)

'User' Class Definition



In [ ]:

    
class User(Document):

    first_name = StringField(max_length=50)
    last_name = StringField(max_length=50)
    email = EmailField(required=True, unique=True)
    
    meta = {
        'db_alias': 'core',
        'collection': 'users',
        
        'indexes': [
            {
                'fields': ['email'], 
                'unique': True,
            },
        ]
    }

'Issue' Class Definition

with embedded documents



In [ ]:

    
class IssueType(EmbeddedDocument):
    name = StringField(required=True, max_length=50)

class Priority(EmbeddedDocument):
    name = StringField(required=True, max_length=50, choices=('Minor','Major','Blocker'))

class Status(EmbeddedDocument):
    name = StringField(required=True, max_length=50)

class Comment(EmbeddedDocument):
    content = StringField()
#     author = ReferenceField(User, reverse_delete_rule=CASCADE)

class Issue(Document):
    issue_type = EmbeddedDocumentField(IssueType)
    priority = EmbeddedDocumentField(Priority)
    status = EmbeddedDocumentField(Status)
    reporter = ReferenceField(User, reverse_delete_rule=CASCADE)  # CASCADE - delete all issues when a user is deleted
    assignee = ReferenceField(User, reverse_delete_rule=CASCADE)

    summary = StringField(max_length=500, required=True)

#     comments = ListField(ReferenceField(Comment))

    tags = ListField(StringField(max_length=30))

    meta = {
        'db_alias': 'core',
        'collection': 'issues',

        'allow_inheritance': True,
        'index_background': True,

        # see: http://docs.mongoengine.org/guide/defining-documents.html?highlight=indexes#indexes
        'indexes': [
            'summary',
            '$summary',  # text index
#             'comments.content',
            # ('reporter','-assignee'), # multi-column
        ]
    }

# class CallIssue(Issue):
#     image_path = StringField()

Create Users

Note: .save() method



In [ ]:

    
cody = User(first_name='Cody', last_name='Fletcher', email='cody@email.com').save()
michael = User(first_name='Michael', last_name='Nguyen', email='michael@email.com').save()

Create Embedded Documents for Issue

Note: NO .save() method required for embedded docs



In [ ]:

    
# Note: NO .save() needed
priority_major = Priority(name='Major')
issue_type_story = IssueType(name='Story')
status_open = Status(name='Open')

Create Issue



In [ ]:

    
issue = Issue(
    priority=priority_major,
    issue_type=issue_type_story,
    status=status_open,
    summary='Test your code',
    reporter=michael,
    assignee=cody,
).save()

# Save the issue id
issue_id = issue.id

Try Out Validation



In [ ]:

    
# Change priority

# This works without raising an exception...
priority_trivial = Priority(name='Trivial')

# ...but this fails validation
issue.priority = priority_trivial
try:
    issue.save()
except ValidationError as e:
    print(f'Exception caught: {e}')

Fetch a Document



In [ ]:

    
issue_fetched = Issue.objects(id=issue_id).first()

if not issue_fetched:
    raise Exception(f'No issue with id {issue_id}')

Clear Database



In [ ]:

    
Issue.drop_collection()
User.drop_collection()



In [ ]: