Bài 11: CRUD Operations

CRUD Là Gì?

CRUD là viết tắt của Create, Read, Update, Delete - bốn operations cơ bản với database.

# CRUD Operations Mapping:# C - Create  → INSERT INTO table# R - Read    → SELECT FROM table# U - Update  → UPDATE table SET# D - Delete  → DELETE FROM table # Django ORM provides methods for all CRUD operations:Post.objects.create()    # CreatePost.objects.get()       # Readpost.save()              # Updatepost.delete()            # Delete

Create Operations

Basic Create - save()

from blog.models import Postfrom django.contrib.auth.models import User # Get authorauthor = User.objects.get(username='admin') # Create instancepost = Post(    title='My First Post',    slug='my-first-post',    content='This is my first blog post.',    author=author,    published=True) # Save to databasepost.save() # Now has primary keyprint(post.id)  # 1print(post.created)  # 2023-10-28 10:00:00 # SQL executed:# INSERT INTO blog_post (title, slug, content, author_id, published, created)# VALUES ('My First Post', 'my-first-post', ..., 1, TRUE, '2023-10-28 10:00:00')

Create with create()

# Create and save in one steppost = Post.objects.create(    title='Second Post',    slug='second-post',    content='Content of second post',    author=author,    published=True) # Immediately saved, has idprint(post.id)  # 2 # Equivalent to:# post = Post(...)# post.save()

Create with Default Values

# Model with defaultsclass Post(models.Model):    title = models.CharField(max_length=200)    slug = models.SlugField(unique=True)    content = models.TextField()    author = models.ForeignKey(User, on_delete=models.CASCADE)    published = models.BooleanField(default=False)  # Default    views = models.IntegerField(default=0)          # Default    created = models.DateTimeField(auto_now_add=True) # Create without specifying defaultspost = Post.objects.create(    title='Draft Post',    slug='draft-post',    content='Draft content',    author=author    # published and views use defaults) print(post.published)  # Falseprint(post.views)      # 0

# ForeignKey relationshipcategory = Category.objects.get(slug='django') post = Post.objects.create(    title='Django Tutorial',    slug='django-tutorial',    content='Learn Django...',    author=author,    category=category  # ForeignKey) # ManyToMany relationshiptag1 = Tag.objects.get(name='python')tag2 = Tag.objects.get(name='web-dev') post.tags.add(tag1, tag2)# Orpost.tags.set([tag1, tag2]) # Create tags and addtutorial_tag = Tag.objects.create(name='tutorial')post.tags.add(tutorial_tag)

Bulk Create

# Create multiple objects efficientlyposts = [    Post(        title=f'Post {i}',        slug=f'post-{i}',        content=f'Content {i}',        author=author    )    for i in range(1, 101)] # Bulk create (single query)Post.objects.bulk_create(posts) # With batch sizePost.objects.bulk_create(posts, batch_size=50) # Bulk create with ignore_conflicts (Django 4.1+)Post.objects.bulk_create(posts, ignore_conflicts=True) # Note: bulk_create() doesn't:# - Call save() method# - Send signals (pre_save, post_save)# - Set auto_now/auto_now_add fields# - Return primary keys (in some databases)

get_or_create()

# Get existing object or create newpost, created = Post.objects.get_or_create(    slug='django-basics',    defaults={        'title': 'Django Basics',        'content': 'Learn Django basics',        'author': author,        'published': True    }) if created:    print("New post created")else:    print("Post already exists") # First call: creates post, created=True# Second call: returns existing post, created=False # Common use case: ensure object existscategory, created = Category.objects.get_or_create(    slug='tutorials',    defaults={'name': 'Tutorials', 'description': 'Tutorial posts'})

update_or_create()

# Update existing or create newpost, created = Post.objects.update_or_create(    slug='django-basics',    defaults={        'title': 'Django Basics - Updated',        'content': 'Updated content',        'published': True    }) # If exists: updates title, content, published# If not exists: creates new with all fields # Use case: sync external dataexternal_posts = fetch_posts_from_api()for post_data in external_posts:    Post.objects.update_or_create(        external_id=post_data['id'],        defaults={            'title': post_data['title'],            'content': post_data['content'],            'updated': timezone.now()        }    )

Read Operations

Get Single Object - get()

# Get by primary keypost = Post.objects.get(pk=1)post = Post.objects.get(id=1) # Get by unique fieldpost = Post.objects.get(slug='django-basics') # Get with multiple conditionspost = Post.objects.get(    slug='django-basics',    author=author) # Exception handlingtry:    post = Post.objects.get(slug='non-existent')except Post.DoesNotExist:    print("Post not found")    post = None try:    post = Post.objects.get(published=True)except Post.MultipleObjectsReturned:    print("Multiple posts found")    post = Post.objects.filter(published=True).first()

Get with get_object_or_404()

from django.shortcuts import get_object_or_404 # In views - raises Http404 if not founddef post_detail(request, slug):    post = get_object_or_404(Post, slug=slug)    return render(request, 'post_detail.html', {'post': post}) # With multiple conditionspost = get_object_or_404(    Post,    slug=slug,    published=True) # Can also use with QuerySetpublished_posts = Post.objects.filter(published=True)post = get_object_or_404(published_posts, slug=slug)

Filter Multiple Objects

# Get all published postspublished_posts = Post.objects.filter(published=True) # Multiple conditions (AND)posts = Post.objects.filter(    published=True,    author=author,    created__year=2023) # Order resultsposts = Post.objects.filter(published=True).order_by('-created') # Limit resultslatest_posts = Post.objects.filter(published=True)[:10] # Countpost_count = Post.objects.filter(published=True).count()

Get All Objects

# Get all postsall_posts = Post.objects.all() # Iteratefor post in all_posts:    print(post.title) # Convert to listposts_list = list(all_posts) # Check if emptyif not all_posts.exists():    print("No posts found")

# Get post with categorypost = Post.objects.get(pk=1)category = post.category  # ForeignKeyprint(category.name) # Reverse relationshipcategory = Category.objects.get(slug='django')posts = category.post_set.all()  # All posts in category # With related_nameclass Post(models.Model):    category = models.ForeignKey(        Category,        on_delete=models.CASCADE,        related_name='posts'  # Custom name    ) category = Category.objects.get(slug='django')posts = category.posts.all()  # Use related_name # ManyToManypost = Post.objects.get(pk=1)tags = post.tags.all()for tag in tags:    print(tag.name)

# Without select_related (N+1 problem)posts = Post.objects.all()for post in posts:  # 1 query    print(post.author.username)  # N queries # With select_related (single JOIN query)posts = Post.objects.select_related('author', 'category')for post in posts:  # 1 query with JOIN    print(post.author.username)  # No additional query    print(post.category.name)    # No additional query # prefetch_related for ManyToManyposts = Post.objects.prefetch_related('tags')for post in posts:  # 2 queries total    for tag in post.tags.all():  # No additional queries        print(tag.name)

Update Operations

Update Single Object

# Method 1: Get, modify, savepost = Post.objects.get(pk=1)post.title = 'Updated Title'post.content = 'Updated content'post.save() # Only update specific fieldspost.title = 'New Title'post.save(update_fields=['title'])# Only title is updated, other fields unchanged # Method 2: Update in one queryPost.objects.filter(pk=1).update(    title='Updated Title',    content='Updated content')# Returns number of rows updated

Update Multiple Objects

# Update all published postsupdated_count = Post.objects.filter(published=True).update(    featured=True)print(f"Updated {updated_count} posts") # Update with conditionPost.objects.filter(views__lt=100).update(    views=100) # Update all objectsPost.objects.all().update(    updated=timezone.now())

Update with F() Expressions

from django.db.models import F # Increment views atomically (no race condition)Post.objects.filter(pk=1).update(    views=F('views') + 1) # Update based on another fieldPost.objects.filter(published=True).update(    popularity=F('views') + F('likes') * 2) # Update in save()post = Post.objects.get(pk=1)post.views = F('views') + 1post.save()post.refresh_from_db()  # Get updated valueprint(post.views)

Bulk Update

# Update multiple objects with different valuesposts = Post.objects.filter(published=False) for post in posts:    post.published = True    post.published_date = timezone.now() # Bulk update (single query per batch)Post.objects.bulk_update(    posts,    ['published', 'published_date'],    batch_size=100) # More efficient than:for post in posts:    post.save()  # Separate query for each

Conditional Updates

from django.db.models import Q, Case, When # Update based on conditionPost.objects.filter(    created__lt=timezone.now() - timezone.timedelta(days=30)).update(archived=True) # Complex conditional updatePost.objects.update(    status=Case(        When(views__gte=1000, then='popular'),        When(views__gte=100, then='normal'),        default='unpopular'    ))

Delete Operations

Delete Single Object

# Method 1: Get then deletepost = Post.objects.get(pk=1)post.delete() # Returns tuple (deleted_count, deleted_objects_dict)# (1, {'blog.Post': 1}) # Method 2: Delete directlyPost.objects.filter(pk=1).delete()

Delete Multiple Objects

# Delete with filterdeleted = Post.objects.filter(published=False).delete()print(f"Deleted {deleted[0]} posts")# (5, {'blog.Post': 5}) # Delete with multiple conditionsPost.objects.filter(    created__year=2020,    views__lt=10).delete() # Delete all (be careful!)Post.objects.all().delete()

Cascade Delete

# on_delete behaviorsclass Post(models.Model):    author = models.ForeignKey(        User,        on_delete=models.CASCADE  # Delete posts when user deleted    )    category = models.ForeignKey(        Category,        on_delete=models.SET_NULL,  # Set to NULL when category deleted        null=True    ) # Delete author → deletes all their postsauthor = User.objects.get(username='john')author.delete()# Cascades to delete all posts by john # Other on_delete options:# - models.PROTECT: Prevent deletion if related objects exist# - models.SET_DEFAULT: Set to default value# - models.SET(value): Set to specific value# - models.DO_NOTHING: Do nothing (may cause integrity error)

Soft Delete Pattern

# Instead of deleting, mark as deletedclass Post(models.Model):    title = models.CharField(max_length=200)    deleted = models.BooleanField(default=False)    deleted_at = models.DateTimeField(null=True, blank=True) # Soft deletepost = Post.objects.get(pk=1)post.deleted = Truepost.deleted_at = timezone.now()post.save() # Custom manager to exclude deletedclass ActiveManager(models.Manager):    def get_queryset(self):        return super().get_queryset().filter(deleted=False) class Post(models.Model):    # ... fields ...    objects = models.Manager()  # All objects    active = ActiveManager()    # Active objects only # UsagePost.objects.all()     # All posts (including deleted)Post.active.all()      # Only active posts

# Delete post and all commentspost = Post.objects.get(pk=1)post.delete()# Automatically deletes related comments (if CASCADE) # Delete only related objectspost = Post.objects.get(pk=1)post.comments.all().delete()  # Delete all comments# Post still exists # Clear ManyToMany relationshipspost.tags.clear()  # Remove all tags from post# Post and tags still exist, just relationship removed

Complete CRUD Examples

Blog Post CRUD

# CREATEdef create_post(request):    if request.method == 'POST':        post = Post.objects.create(            title=request.POST['title'],            slug=slugify(request.POST['title']),            content=request.POST['content'],            author=request.user,            published=True        )        return redirect('post_detail', slug=post.slug)    return render(request, 'create_post.html') # READdef post_detail(request, slug):    post = get_object_or_404(Post, slug=slug)    # Increment views    Post.objects.filter(pk=post.pk).update(views=F('views') + 1)    post.refresh_from_db()    return render(request, 'post_detail.html', {'post': post}) def post_list(request):    posts = Post.objects.filter(published=True).order_by('-created')    return render(request, 'post_list.html', {'posts': posts}) # UPDATEdef update_post(request, slug):    post = get_object_or_404(Post, slug=slug, author=request.user)        if request.method == 'POST':        post.title = request.POST['title']        post.content = request.POST['content']        post.save(update_fields=['title', 'content', 'updated'])        return redirect('post_detail', slug=post.slug)        return render(request, 'update_post.html', {'post': post}) # DELETEdef delete_post(request, slug):    post = get_object_or_404(Post, slug=slug, author=request.user)        if request.method == 'POST':        post.delete()        return redirect('post_list')        return render(request, 'confirm_delete.html', {'post': post})

API-Style CRUD

from django.http import JsonResponsefrom django.views.decorators.csrf import csrf_exemptimport json # CREATE@csrf_exemptdef api_create_post(request):    if request.method == 'POST':        data = json.loads(request.body)        post = Post.objects.create(            title=data['title'],            content=data['content'],            author=request.user        )        return JsonResponse({            'id': post.id,            'title': post.title,            'created': post.created.isoformat()        }, status=201) # READdef api_get_post(request, pk):    try:        post = Post.objects.get(pk=pk)        return JsonResponse({            'id': post.id,            'title': post.title,            'content': post.content,            'author': post.author.username,            'created': post.created.isoformat()        })    except Post.DoesNotExist:        return JsonResponse({'error': 'Post not found'}, status=404) # UPDATE@csrf_exemptdef api_update_post(request, pk):    if request.method == 'PUT':        try:            post = Post.objects.get(pk=pk)            data = json.loads(request.body)                        post.title = data.get('title', post.title)            post.content = data.get('content', post.content)            post.save()                        return JsonResponse({'message': 'Post updated'})        except Post.DoesNotExist:            return JsonResponse({'error': 'Post not found'}, status=404) # DELETE@csrf_exemptdef api_delete_post(request, pk):    if request.method == 'DELETE':        try:            post = Post.objects.get(pk=pk)            post.delete()            return JsonResponse({'message': 'Post deleted'})        except Post.DoesNotExist:            return JsonResponse({'error': 'Post not found'}, status=404)

Bài Tập

Bài 1: Basic CRUD

Task: Implement CRUD for Comment model:

# Comment model:class Comment(models.Model):    post = models.ForeignKey(Post, on_delete=models.CASCADE)    author = models.CharField(max_length=100)    email = models.EmailField()    content = models.TextField()    approved = models.BooleanField(default=False)    created = models.DateTimeField(auto_now_add=True) # Implement:# 1. Create comment for a post# 2. Get all comments for a post# 3. Get single comment by id# 4. Update comment content# 5. Delete comment# 6. Approve comment (set approved=True)

Bài 2: Bulk Operations

Task: Practice bulk operations:

# 1. Create 50 posts using bulk_create# 2. Update all unpublished posts to published# 3. Increment views by 10 for all posts# 4. Delete all posts with views < 100# 5. Soft delete posts older than 1 year

Bài 3: get_or_create Pattern

Task: Use get_or_create for tags:

# Create function to get or create tagsdef get_or_create_tags(tag_names):    """    Takes list of tag names.    Returns list of Tag objects (created or existing).        Example:    tags = get_or_create_tags(['python', 'django', 'web'])    """    # Implement here    pass # Use it to add tags to postpost = Post.objects.get(pk=1)tags = get_or_create_tags(['python', 'django', 'tutorial'])post.tags.set(tags)

Bài 4: Complete CRUD Views

Task: Create complete CRUD views for Article:

# Implement these views:# 1. article_list(request) - Show all published articles# 2. article_detail(request, slug) - Show single article, increment views# 3. article_create(request) - Create new article (requires login)# 4. article_update(request, slug) - Update article (only author)# 5. article_delete(request, slug) - Delete article (only author)# 6. my_articles(request) - Show current user's articles # Include:# - Permission checks (author only for update/delete)# - Error handling (DoesNotExist, Http404)# - Redirects after create/update/delete# - Messages for success/error

Tài Liệu Tham Khảo

Previous: Bài 10: QuerySets | Next: Bài 12: Forms

#create

Django CRUD Operations