1from akron import Akron
2from pydantic import BaseModel
3from akron.models import ModelMixin
4from datetime import datetime
5from typing import List, Dict, Optional
6from bson import ObjectId
7
8# Define your model for MongoDB documents
9class BlogPost(BaseModel, ModelMixin):
10 _id: Optional[str] = None # MongoDB ObjectId
11 title: str
12 content: str
13 author: Dict[str, str]
14 tags: List[str]
15 metadata: Dict[str, any]
16 comments: List[Dict[str, any]] = []
17 published: bool = False
18 view_count: int = 0
19 created_at: datetime
20 updated_at: Optional[datetime] = None
21
22# Connect to MongoDB
23db = Akron("mongodb://localhost:27017/blog")
24
25# Note: MongoDB doesn't require explicit table creation
26# Collections are created automatically when first document is inserted
27
28# Insert blog posts
29post1 = BlogPost(
30 title="Getting Started with MongoDB",
31 content="MongoDB is a document database...",
32 author={
33 "name": "Alice Johnson",
34 "email": "alice@example.com",
35 "bio": "Database enthusiast"
36 },
37 tags=["mongodb", "nosql", "database", "tutorial"],
38 metadata={
39 "category": "tutorial",
40 "difficulty": "beginner",
41 "estimated_read_time": 8,
42 "featured": True
43 },
44 created_at=datetime.now()
45)
46
47post2 = BlogPost(
48 title="Advanced MongoDB Aggregation",
49 content="Learn complex aggregation pipelines...",
50 author={
51 "name": "Bob Smith",
52 "email": "bob@example.com",
53 "bio": "Senior Developer"
54 },
55 tags=["mongodb", "aggregation", "advanced"],
56 metadata={
57 "category": "advanced",
58 "difficulty": "expert",
59 "estimated_read_time": 20
60 },
61 comments=[
62 {
63 "author": "Jane Doe",
64 "content": "Excellent tutorial!",
65 "created_at": datetime.now()
66 }
67 ],
68 published=True,
69 view_count=342,
70 created_at=datetime.now()
71)
72
73# Insert documents
74BlogPost.insert(db, post1)
75BlogPost.insert(db, post2)
76
77# Query documents
78all_posts = BlogPost.select(db)
79print(f"Total blog posts: {len(all_posts)}")
80
81# Find published posts
82published_posts = BlogPost.select(db, where={"published": True})
83print(f"Published posts: {len(published_posts)}")
84
85# Complex queries with MongoDB operators
86# Find posts by specific author
87alice_posts = BlogPost.select(db, where={"author.name": "Alice Johnson"})
88
89# Find posts with specific tags
90tutorial_posts = BlogPost.select(db, where={"tags": {"$in": ["tutorial"]}})
91
92# Find posts with high view count
93popular_posts = BlogPost.select(db, where={"view_count": {"$gte": 100}})
94
95# Text search (requires text index)
96# db.execute_raw("db.blogposts.createIndex({title: 'text', content: 'text'})")
97# search_results = BlogPost.select(db, where={"$text": {"$search": "mongodb tutorial"}})
98
99# Update documents
100BlogPost.update(
101 db,
102 {"title": "Getting Started with MongoDB"},
103 {
104 "published": True,
105 "view_count": 45,
106 "updated_at": datetime.now()
107 }
108)
109
110# Add comment to a post
111BlogPost.update(
112 db,
113 {"title": "Advanced MongoDB Aggregation"},
114 {
115 "$push": {
116 "comments": {
117 "author": "Charlie Brown",
118 "content": "Very detailed explanation!",
119 "created_at": datetime.now()
120 }
121 }
122 }
123)
124
125# Increment view count
126BlogPost.update(
127 db,
128 {"title": "Getting Started with MongoDB"},
129 {"$inc": {"view_count": 1}}
130)
131
132print("MongoDB operations completed successfully!")