Showing posts with label database. Show all posts
Showing posts with label database. Show all posts

Monday, 3 November 2025

Building a Type-Safe Full-Stack Application with tRPC, Next.js, and Prisma (2025 Guide)

November 03, 2025 0

Building a Type-Safe Full-Stack Application with tRPC, Next.js, and Prisma

Type-safe full-stack architecture with tRPC, Next.js, and Prisma showing end-to-end TypeScript type flow

In 2025, type safety has evolved from a development luxury to a production necessity. The combination of tRPC, Next.js, and Prisma represents the pinnacle of full-stack type safety, enabling developers to build robust applications with end-to-end TypeScript coverage. This comprehensive guide explores how to create a completely type-safe full-stack application where your frontend, backend, and database schema are seamlessly connected through automatic type inference. Whether you're building a SaaS platform, e-commerce site, or internal tool, mastering this stack will eliminate entire classes of runtime errors and dramatically accelerate your development velocity.

🚀 Why End-to-End Type Safety Matters in 2025

Traditional full-stack development often suffers from type mismatches between frontend and backend, leading to runtime errors and development friction. The tRPC + Next.js + Prisma stack solves this by creating a unified type system that spans your entire application.

  • Zero API Contracts: Automatic type sharing between frontend and backend
  • Development Speed: Instant feedback and autocomplete across the entire stack
  • Runtime Safety: Catch errors at compile time rather than in production
  • Maintainability: Refactor with confidence across frontend and backend
  • Developer Experience: Superior IDE support and documentation

💻 Complete Project Setup and Configuration

Let's start with a complete project setup that establishes our type-safe foundation.

// package.json - Complete dependencies
{
  "name": "type-safe-fullstack",
  "version": "1.0.0",
  "scripts": {
    "dev": "next dev",
    "build": "next build",
    "start": "next start",
    "db:generate": "prisma generate",
    "db:push": "prisma db push",
    "db:studio": "prisma studio",
    "type-check": "tsc --noEmit"
  },
  "dependencies": {
    "@prisma/client": "^5.6.0",
    "@tanstack/react-query": "^5.0.0",
    "@trpc/client": "^11.0.0",
    "@trpc/next": "^11.0.0",
    "@trpc/react-query": "^11.0.0",
    "@trpc/server": "^11.0.0",
    "next": "14.0.0",
    "react": "^18.2.0",
    "react-dom": "^18.2.0",
    "superjson": "^2.0.0",
    "zod": "^3.22.0"
  },
  "devDependencies": {
    "@types/node": "^20.0.0",
    "@types/react": "^18.2.0",
    "@types/react-dom": "^18.2.0",
    "prisma": "^5.6.0",
    "typescript": "^5.2.0"
  }
}

// tsconfig.json - Strict TypeScript configuration
{
  "compilerOptions": {
    "target": "es5",
    "lib": ["dom", "dom.iterable", "es6"],
    "allowJs": true,
    "skipLibCheck": true,
    "strict": true,
    "noEmit": true,
    "esModuleInterop": true,
    "module": "esnext",
    "moduleResolution": "bundler",
    "resolveJsonModule": true,
    "isolatedModules": true,
    "jsx": "preserve",
    "incremental": true,
    "plugins": [
      {
        "name": "next"
      }
    ],
    "baseUrl": ".",
    "paths": {
      "@/*": ["./src/*"],
      "~/*": ["./*"]
    }
  },
  "include": ["next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts"],
  "exclude": ["node_modules"]
}
  

🛠️ tRPC Setup with Advanced Configuration

Setting up tRPC correctly is crucial for type safety. Here's a complete configuration with error handling and middleware.

// src/server/trpc.ts - tRPC configuration
import { initTRPC, TRPCError } from '@trpc/server';
import { type CreateNextContextOptions } from '@trpc/server/adapters/next';
import superjson from 'superjson';
import { ZodError } from 'zod';
import { prisma } from './prisma';

// Context creation
export const createTRPCContext = (opts: CreateNextContextOptions) => {
  return {
    prisma,
    req: opts.req,
    res: opts.res,
    user: null, // Would come from auth in real app
  };
};

// Initialize tRPC
const t = initTRPC.context<typeof createTRPCContext>().create({
  transformer: superjson,
  errorFormatter({ shape, error }) {
    return {
      ...shape,
      data: {
        ...shape.data,
        zodError:
          error.cause instanceof ZodError ? error.cause.flatten() : null,
      },
    };
  },
});

// Middlewares
export const createTRPCRouter = t.router;
export const publicProcedure = t.procedure;

// Authentication middleware
const isAuthed = t.middleware(({ ctx, next }) => {
  if (!ctx.user) {
    throw new TRPCError({ code: 'UNAUTHORIZED' });
  }
  return next({
    ctx: {
      ...ctx,
      user: ctx.user, // user is now non-null
    },
  });
});

export const protectedProcedure = t.procedure.use(isAuthed);
  

🚀 Next.js API Route Configuration

Setting up the tRPC API route in Next.js to handle both HTTP and WebSocket requests.

// src/pages/api/trpc/[trpc].ts - tRPC API handler
import { createNextApiHandler } from '@trpc/server/adapters/next';
import { appRouter } from '../../../server/routes/_app';
import { createTRPCContext } from '../../../server/trpc';

export default createNextApiHandler({
  router: appRouter,
  createContext: createTRPCContext,
  onError:
    process.env.NODE_ENV === 'development'
      ? ({ path, error }) => {
          console.error(
            `❌ tRPC failed on ${path ?? '<no-path>'}: ${error.message}`
          );
        }
      : undefined,
  responseMeta({ ctx, paths, type, errors }) {
    // Cache API responses for 1 minute
    const allPublic = paths && paths.every((path) => path.includes('public'));
    const allOk = errors.length === 0;
    const isQuery = type === 'query';

    if (ctx?.res && allPublic && allOk && isQuery) {
      return {
        headers: {
          'cache-control': `s-maxage=60, stale-while-revalidate=300`,
        },
      };
    }
    return {};
  },
});

// src/utils/trpc.ts - Frontend tRPC client
import { httpBatchLink, loggerLink } from '@trpc/client';
import { createTRPCNext } from '@trpc/next';
import superjson from 'superjson';
import { type AppRouter } from '../server/routes/_app';

function getBaseUrl() {
  if (typeof window !== 'undefined') return '';
  if (process.env.VERCEL_URL) return `https://${process.env.VERCEL_URL}`;
  return `http://localhost:${process.env.PORT ?? 3000}`;
}

export const trpc = createTRPCNext<AppRouter>({
  config() {
    return {
      transformer: superjson,
      links: [
        loggerLink({
          enabled: (opts) =>
            process.env.NODE_ENV === 'development' ||
            (opts.direction === 'down' && opts.result instanceof Error),
        }),
        httpBatchLink({
          url: `${getBaseUrl()}/api/trpc`,
        }),
      ],
    };
  },
  ssr: true,
});
  

🎯 Type-Safe Frontend Components

Leveraging tRPC's type inference to build completely type-safe React components.

// src/components/PostList.tsx - Type-safe post listing
import { useState } from 'react';
import { trpc } from '../utils/trpc';

export function PostList() {
  const [search, setSearch] = useState('');
  const {
    data,
    fetchNextPage,
    hasNextPage,
    isFetchingNextPage,
    status,
    error,
  } = trpc.post.list.useInfiniteQuery(
    {
      limit: 10,
      search: search || undefined,
    },
    {
      getNextPageParam: (lastPage) => lastPage.nextCursor,
      staleTime: 5 * 60 * 1000, // 5 minutes
    }
  );

  if (status === 'loading') {
    return <div>Loading posts...</div>;
  }

  if (status === 'error') {
    return <div>Error: {error.message}</div>;
  }

  return (
    <div className="space-y-6">
      <div className="flex gap-4">
        <input
          type="text"
          placeholder="Search posts..."
          value={search}
          onChange={(e) => setSearch(e.target.value)}
          className="px-4 py-2 border rounded-lg"
        />
      </div>

      <div className="space-y-4">
        {data.pages.map((page, pageIndex) => (
          <div key={pageIndex} className="space-y-4">
            {page.posts.map((post) => (
              <PostCard key={post.id} post={post} />
            ))}
          </div>
        ))}
      </div>

      {hasNextPage && (
        <button
          onClick={() => fetchNextPage()}
          disabled={isFetchingNextPage}
          className="px-4 py-2 bg-blue-500 text-white rounded-lg disabled:opacity-50"
        >
          {isFetchingNextPage ? 'Loading more...' : 'Load More'}
        </button>
      )}
    </div>
  );
}

// src/components/CreatePostForm.tsx - Type-safe form with validation
import { useForm } from 'react-hook-form';
import { zodResolver } from '@hookform/resolvers/zod';
import { z } from 'zod';
import { trpc } from '../utils/trpc';

const createPostSchema = z.object({
  title: z.string().min(1, 'Title is required').max(255),
  content: z.string().optional(),
  tagIds: z.array(z.string()).optional(),
});

type CreatePostInput = z.infer<typeof createPostSchema>;

export function CreatePostForm() {
  const utils = trpc.useContext();
  const { data: tags } = trpc.tag.list.useQuery();
  
  const createPost = trpc.post.create.useMutation({
    onSuccess: () => {
      utils.post.list.invalidate();
      reset();
    },
  });

  const {
    register,
    handleSubmit,
    formState: { errors },
    reset,
  } = useForm<CreatePostInput>({
    resolver: zodResolver(createPostSchema),
  });

  const onSubmit = (data: CreatePostInput) => {
    createPost.mutate(data);
  };

  return (
    <form onSubmit={handleSubmit(onSubmit)} className="space-y-4 p-6 border rounded-lg">
      <h2 className="text-lg font-semibold">Create New Post</h2>
      
      <div>
        <label className="block text-sm font-medium mb-1">Title</label>
        <input
          {...register('title')}
          className="w-full px-3 py-2 border rounded-lg"
          placeholder="Post title"
        />
        {errors.title && (
          <p className="text-red-500 text-sm mt-1">{errors.title.message}</p>
        )}
      </div>

      <button
        type="submit"
        disabled={createPost.isLoading}
        className="px-4 py-2 bg-blue-500 text-white rounded-lg disabled:opacity-50"
      >
        {createPost.isLoading ? 'Creating...' : 'Create Post'}
      </button>
    </form>
  );
}
  

🔒 Advanced Patterns and Best Practices

Beyond the basics, these advanced patterns will make your type-safe application production-ready.

  • Error Handling: Structured error types and client-side error boundaries
  • Authentication: Type-safe session management and protected procedures
  • Caching Strategies: Optimistic updates and query invalidation
  • Testing: End-to-end type-safe testing utilities
  • Performance: Code splitting and bundle optimization

⚡ Key Takeaways

  1. End-to-End Type Safety: Automatic type sharing eliminates API contract mismatches
  2. Development Velocity: Instant feedback and autocomplete across the entire stack
  3. Runtime Confidence: Compile-time error catching prevents production issues
  4. Maintainability: Refactoring becomes safe and predictable
  5. Developer Experience: Superior IDE support reduces cognitive load
  6. Performance: Built-in optimizations like batching and caching
  7. Scalability: Modular router structure supports growing applications

❓ Frequently Asked Questions

How does tRPC compare to GraphQL or REST APIs?
tRPC provides automatic type safety without the complexity of GraphQL schemas or the manual type definitions of REST. It's ideal for TypeScript-focused teams building full-stack applications where you control both frontend and backend. GraphQL excels at public APIs and complex data requirements, while REST remains the universal standard for web APIs.
Can I use tRPC with existing REST APIs or databases?
Yes, tRPC can coexist with existing APIs. You can gradually migrate endpoints to tRPC or use it for new features while maintaining existing REST APIs. For databases, Prisma supports most major databases, and you can use tRPC with any data source by creating custom procedures that don't rely on Prisma.
What about authentication and authorization in tRPC?
tRPC supports authentication through middleware. You can create protected procedures that require authentication and access user context in your resolvers. The type system ensures that protected procedures can only be called with proper authentication, and user data is type-safe throughout your application.
How do I handle file uploads with tRPC?
While tRPC works best with JSON data, you can handle file uploads by using Next.js API routes for file handling and tRPC for metadata. Alternatively, use base64 encoding for small files or create a separate file upload service that integrates with your tRPC API through procedure calls.
Is tRPC suitable for large-scale production applications?
Absolutely. tRPC is used in production by companies like Cal.com, Ping.gg, and others. It scales well through router composition, middleware chains, and proper architecture. The type safety actually becomes more valuable as the application grows, preventing entire classes of errors in large codebases.
How do I deploy a tRPC + Next.js application?
Deployment is straightforward with platforms like Vercel, Netlify, or any Node.js hosting provider. Since it's a standard Next.js application, you get all the benefits of Next.js deployment including automatic API route handling, static generation, and server-side rendering. Just ensure your database connections are properly configured for your deployment environment.

💬 Have you built applications with tRPC, Next.js, and Prisma? Share your experiences, challenges, or tips in the comments below! If you found this guide helpful, please share it with your team or on social media to help others master type-safe full-stack development.

About LK-TECH Academy — Practical tutorials & explainers on software engineering, AI, and infrastructure. Follow for concise, hands-on guides.

Wednesday, 29 October 2025

Vector Database Deep Dive: Building a Semantic Search Engine with Weaviate (2025 Guide)

October 29, 2025 0

Vector Database Deep Dive: Building a Semantic Search Engine with Weaviate

weaviate-vector-database-semantic-search-architecture-2025

In 2025, semantic search has evolved from a nice-to-have feature to a fundamental requirement for modern applications. While traditional keyword-based search struggles with context and meaning, vector databases like Weaviate enable true understanding through mathematical representations of meaning. This comprehensive guide explores how to build production-ready semantic search systems using Weaviate, covering everything from vector embeddings and similarity algorithms to hybrid search patterns and real-time updates. Whether you're building an intelligent document retrieval system, product recommendation engine, or AI-powered knowledge base, mastering vector databases will transform how you handle unstructured data.

🚀 Why Vector Databases Dominate AI Applications in 2025

Vector databases have become the backbone of modern AI systems by enabling efficient storage and retrieval of high-dimensional embeddings. Unlike traditional databases that match exact values, vector databases find semantically similar content, making them perfect for AI applications that need to understand context and meaning.

  • Semantic Understanding: Find content based on meaning rather than exact keyword matches
  • Multi-modal Search: Search across text, images, audio, and video using the same interface
  • Real-time Performance: Handle millions of vectors with sub-second query times
  • Hybrid Capabilities: Combine vector search with traditional filtering and keyword search
  • Scalability: Scale horizontally to handle growing datasets and query loads

🔧 Weaviate Architecture: Understanding the Core Components

Weaviate's modular architecture separates storage, computation, and embedding generation, providing flexibility and performance. Understanding these components is crucial for building efficient systems.

  • Vector Index: HNSW algorithm for efficient approximate nearest neighbor search
  • Modules System: Pluggable components for embeddings, text processing, and more
  • GraphQL Interface: Unified query language for both vector and traditional operations
  • Replication & Sharding: Built-in high availability and horizontal scaling
  • Multi-tenancy: Isolated data partitions for different applications or customers

💻 Complete Weaviate Setup and Configuration

Let's start with a complete Docker-based Weaviate setup with custom modules and optimized configuration for production use.


# docker-compose.yml - Production Weaviate Setup
version: '3.4'
services:
  weaviate:
    command:
    - --host
    - 0.0.0.0
    - --port
    - '8080'
    - --scheme
    - http
    image: cr.weaviate.io/semitechnologies/weaviate:1.24.0
    ports:
    - 8080:8080
    - 50051:50051
    restart: on-failure:0
    environment:
      OPENAI_APIKEY: ${OPENAI_APIKEY}
      QUERY_DEFAULTS_LIMIT: 25
      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
      PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
      DEFAULT_VECTORIZER_MODULE: 'text2vec-openai'
      ENABLE_MODULES: 'text2vec-openai,generative-openai,qna-openai'
      CLUSTER_HOSTNAME: 'node1'
      LOG_LEVEL: 'info'
      MAX_IMPORT_BATCH_SIZE: '100'
      MAX_IMPORT_CONCURRENT_REQUESTS: '4'
    volumes:
      - weaviate_data:/var/lib/weaviate
    deploy:
      resources:
        limits:
          memory: 8G
        reservations:
          memory: 4G

  # Optional: Add monitoring with Prometheus
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
      - '--web.console.libraries=/etc/prometheus/console_libraries'
      - '--web.console.templates=/etc/prometheus/consoles'
      - '--storage.tsdb.retention.time=200h'
      - '--web.enable-lifecycle'

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - ./monitoring/grafana/dashboards:/var/lib/grafana/dashboards
      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning
    depends_on:
      - prometheus

volumes:
  weaviate_data:

  

🛠️ Python Client Implementation and Schema Design

Proper schema design is crucial for performance and functionality. Here's how to define classes and properties with optimal vectorization settings.


# weaviate_schema.py - Advanced Schema Design
import weaviate
from weaviate.classes.config import Configure, Property, DataType
from weaviate.classes.init import AdditionalConfig, Timeout
import asyncio

class WeaviateSemanticSearch:
    def __init__(self, endpoint="http://localhost:8080"):
        self.client = weaviate.WeaviateClient(
            additional_config=AdditionalConfig(
                timeout=Timeout(init=60, query=300, insert=120),
                grpc_port_experimental=50051
            )
        )
        self.client.connect_to_local()
    
    async def create_document_schema(self):
        """Create optimized schema for document search"""
        document_class = {
            "class": "Document",
            "description": "A document for semantic search",
            "vectorizer": "text2vec-openai",
            "moduleConfig": {
                "text2vec-openai": {
                    "model": "text-embedding-3-large",
                    "modelVersion": "latest",
                    "type": "text",
                    "vectorizeClassName": False
                },
                "generative-openai": {
                    "model": "gpt-4"
                }
            },
            "properties": [
                {
                    "name": "title",
                    "dataType": ["text"],
                    "description": "Document title",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": False,
                            "vectorizePropertyName": False
                        }
                    }
                },
                {
                    "name": "content",
                    "dataType": ["text"],
                    "description": "Document content",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": False,
                            "vectorizePropertyName": False
                        }
                    }
                },
                {
                    "name": "category",
                    "dataType": ["text"],
                    "description": "Document category",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": True,
                            "vectorizePropertyName": False
                        }
                    }
                },
                {
                    "name": "tags",
                    "dataType": ["text[]"],
                    "description": "Document tags",
                    "moduleConfig": {
                        "text2vec-openai": {
                            "skip": True,
                            "vectorizePropertyName": False
                        }
                    }
                },
                {
                    "name": "created_at",
                    "dataType": ["date"],
                    "description": "Creation timestamp"
                },
                {
                    "name": "updated_at",
                    "dataType": ["date"],
                    "description": "Last update timestamp"
                },
                {
                    "name": "author",
                    "dataType": ["text"],
                    "description": "Document author"
                },
                {
                    "name": "word_count",
                    "dataType": ["int"],
                    "description": "Number of words in document"
                },
                {
                    "name": "readability_score",
                    "dataType": ["number"],
                    "description": "Document readability score"
                }
            ],
            "vectorIndexType": "hnsw",
            "vectorIndexConfig": {
                "distance": "cosine",
                "ef": 128,
                "efConstruction": 128,
                "maxConnections": 32,
                "cleanupIntervalSeconds": 300,
                "dynamicEfMin": 100,
                "dynamicEfMax": 500,
                "dynamicEfFactor": 8
            },
            "shardingConfig": {
                "desiredCount": 3,
                "actualCount": 3,
                "virtualPerPhysical": 128,
                "key": "_id",
                "strategy": "hash",
                "function": "murmur3"
            },
            "replicationConfig": {
                "factor": 2,
                "asyncEnabled": False
            }
        }
        
        # Create the class
        try:
            self.client.collections.create_from_dict(document_class)
            print("Document schema created successfully")
        except Exception as e:
            print(f"Schema creation error: {e}")
    
    async def create_multimodal_schema(self):
        """Create schema for multi-modal search (text + images)"""
        multimodal_class = {
            "class": "MultimodalContent",
            "description": "Content with both text and images",
            "vectorizer": "text2vec-openai",
            "moduleConfig": {
                "text2vec-openai": {
                    "model": "text-embedding-3-large",
                    "vectorizeClassName": False
                }
            },
            "properties": [
                {
                    "name": "text_content",
                    "dataType": ["text"],
                    "description": "Text content for vectorization"
                },
                {
                    "name": "image_url",
                    "dataType": ["text"],
                    "description": "URL to image file"
                },
                {
                    "name": "image_embedding",
                    "dataType": ["blob"],
                    "description": "Pre-computed image embeddings"
                },
                {
                    "name": "content_type",
                    "dataType": ["text"],
                    "description": "Type of content (text, image, mixed)"
                }
            ]
        }
        
        self.client.collections.create_from_dict(multimodal_class)
        print("Multimodal schema created successfully")
    
    def get_collection(self, class_name):
        """Get collection with proper configuration"""
        return self.client.collections.get(
            class_name,
            consistency_level=weaviate.classes.config.ConsistencyLevel.QUORUM
        )

# Initialize and create schemas
async def main():
    search_engine = WeaviateSemanticSearch()
    await search_engine.create_document_schema()
    await search_engine.create_multimodal_schema()

if __name__ == "__main__":
    asyncio.run(main())

  

🚀 Advanced Data Ingestion and Vectorization

Efficient data ingestion is critical for performance. Here's how to implement batch processing, error handling, and custom vectorization.


# data_ingestion.py - Advanced Data Import
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime
import uuid
from typing import List, Dict, Any
import logging

class WeaviateDataManager:
    def __init__(self, client):
        self.client = client
        self.logger = logging.getLogger(__name__)
        
    async def import_documents_batch(self, documents: List[Dict], batch_size: int = 100):
        """Import documents with batch processing and error handling"""
        collection = self.client.collections.get("Document")
        
        successful_imports = 0
        failed_imports = 0
        
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            batch_objects = []
            
            for doc in batch:
                try:
                    # Validate required fields
                    if not doc.get('title') or not doc.get('content'):
                        self.logger.warning(f"Skipping document missing required fields: {doc.get('id', 'unknown')}")
                        failed_imports += 1
                        continue
                    
                    # Create Weaviate object
                    weaviate_obj = {
                        "title": doc['title'],
                        "content": doc['content'],
                        "category": doc.get('category', 'general'),
                        "tags": doc.get('tags', []),
                        "created_at": doc.get('created_at', datetime.now().isoformat()),
                        "updated_at": doc.get('updated_at', datetime.now().isoformat()),
                        "author": doc.get('author', 'unknown'),
                        "word_count": doc.get('word_count', len(doc['content'].split())),
                        "readability_score": doc.get('readability_score', 0.0)
                    }
                    
                    # Add UUID if provided, otherwise generate
                    if 'id' in doc:
                        weaviate_obj['id'] = doc['id']
                    
                    batch_objects.append(weaviate_obj)
                    
                except Exception as e:
                    self.logger.error(f"Error processing document: {e}")
                    failed_imports += 1
            
            if batch_objects:
                try:
                    # Insert batch with retry logic
                    result = await self._insert_with_retry(collection, batch_objects)
                    successful_imports += len(result.successful)
                    failed_imports += len(result.failed) if hasattr(result, 'failed') else 0
                    
                    self.logger.info(f"Batch {i//batch_size + 1}: {len(result.successful)} successful, {len(result.failed) if hasattr(result, 'failed') else 0} failed")
                    
                except Exception as e:
                    self.logger.error(f"Batch insert failed: {e}")
                    failed_imports += len(batch_objects)
            
            # Rate limiting to avoid overwhelming the server
            await asyncio.sleep(0.1)
        
        return {
            "successful": successful_imports,
            "failed": failed_imports,
            "total": len(documents)
        }
    
    async def _insert_with_retry(self, collection, objects, max_retries=3):
        """Insert with exponential backoff retry logic"""
        for attempt in range(max_retries):
            try:
                return collection.data.insert_many(objects)
            except Exception as e:
                if attempt == max_retries - 1:
                    raise e
                wait_time = (2 ** attempt) + 1
                self.logger.warning(f"Insert failed, retrying in {wait_time}s: {e}")
                await asyncio.sleep(wait_time)
    
    async def import_from_csv(self, csv_path: str, **kwargs):
        """Import documents from CSV file"""
        df = pd.read_csv(csv_path)
        documents = []
        
        for _, row in df.iterrows():
            doc = {
                'title': row.get('title', ''),
                'content': row.get('content', ''),
                'category': row.get('category', 'general'),
                'tags': row.get('tags', '').split(';') if pd.notna(row.get('tags')) else [],
                'author': row.get('author', 'unknown'),
                'word_count': row.get('word_count', 0),
                'readability_score': row.get('readability_score', 0.0)
            }
            
            # Add timestamp if available
            if 'created_at' in row and pd.notna(row['created_at']):
                doc['created_at'] = row['created_at']
            
            documents.append(doc)
        
        return await self.import_documents_batch(documents, **kwargs)
    
    async def update_document(self, doc_id: str, updates: Dict[str, Any]):
        """Update existing document with partial updates"""
        collection = self.client.collections.get("Document")
        
        try:
            # Get existing document
            existing = collection.query.fetch_object_by_id(doc_id)
            if not existing:
                raise ValueError(f"Document {doc_id} not found")
            
            # Merge updates
            updated_data = {**existing.properties, **updates}
            updated_data['updated_at'] = datetime.now().isoformat()
            
            # Update document
            collection.data.update(
                uuid=doc_id,
                properties=updated_data
            )
            
            self.logger.info(f"Document {doc_id} updated successfully")
            return True
            
        except Exception as e:
            self.logger.error(f"Failed to update document {doc_id}: {e}")
            return False
    
    async def delete_documents_by_filter(self, filters: Dict[str, Any]):
        """Delete documents matching filter criteria"""
        collection = self.client.collections.get("Document")
        
        try:
            # Build GraphQL where filter
            where_clause = self._build_where_clause(filters)
            
            # Execute batch delete
            result = collection.data.delete_many(where=where_clause)
            
            self.logger.info(f"Deleted {result} documents matching filters")
            return result
            
        except Exception as e:
            self.logger.error(f"Failed to delete documents: {e}")
            return 0
    
    def _build_where_clause(self, filters: Dict[str, Any]) -> Dict[str, Any]:
        """Build GraphQL where clause from filters"""
        where_conditions = []
        
        for field, value in filters.items():
            if isinstance(value, (list, tuple)):
                where_conditions.append({
                    "path": [field],
                    "operator": "ContainsAny",
                    "valueText": value
                })
            elif isinstance(value, str):
                where_conditions.append({
                    "path": [field],
                    "operator": "Equal",
                    "valueText": value
                })
            elif isinstance(value, (int, float)):
                where_conditions.append({
                    "path": [field],
                    "operator": "Equal",
                    "valueNumber": value
                })
            elif isinstance(value, dict):
                # Handle range queries
                if 'min' in value and 'max' in value:
                    where_conditions.append({
                        "path": [field],
                        "operator": "And",
                        "operands": [
                            {"path": [field], "operator": "GreaterThanEqual", "valueNumber": value['min']},
                            {"path": [field], "operator": "LessThanEqual", "valueNumber": value['max']}
                        ]
                    })
        
        if len(where_conditions) == 1:
            return where_conditions[0]
        else:
            return {
                "operator": "And",
                "operands": where_conditions
            }

# Example usage
async def example_import():
    from weaviate_schema import WeaviateSemanticSearch
    
    search_engine = WeaviateSemanticSearch()
    data_manager = WeaviateDataManager(search_engine.client)
    
    # Sample documents
    sample_docs = [
        {
            "title": "Introduction to Machine Learning",
            "content": "Machine learning is a subset of artificial intelligence that enables computers to learn without being explicitly programmed.",
            "category": "AI",
            "tags": ["machine-learning", "ai", "tutorial"],
            "author": "AI Researcher",
            "word_count": 150,
            "readability_score": 8.5
        },
        {
            "title": "Deep Learning Fundamentals",
            "content": "Deep learning uses neural networks with multiple layers to learn complex patterns in large datasets.",
            "category": "AI",
            "tags": ["deep-learning", "neural-networks", "advanced"],
            "author": "ML Engineer",
            "word_count": 200,
            "readability_score": 7.8
        }
    ]
    
    result = await data_manager.import_documents_batch(sample_docs)
    print(f"Import result: {result}")

if __name__ == "__main__":
    asyncio.run(example_import())

  

🔍 Advanced Query Patterns and Semantic Search

Weaviate's GraphQL interface enables powerful query patterns. Here are advanced search techniques for production systems.


# semantic_search.py - Advanced Query Patterns
import weaviate
from weaviate.classes.query import Filter, HybridFusion
from typing import List, Dict, Any, Optional
import numpy as np

class AdvancedSemanticSearch:
    def __init__(self, client):
        self.client = client
        self.collection = client.collections.get("Document")
    
    async def semantic_search(self, query: str, limit: int = 10, 
                           filters: Optional[Dict] = None,
                           certainty: float = 0.7):
        """Basic semantic search with filters"""
        response = self.collection.query.near_text(
            query=query,
            limit=limit,
            filters=self._build_filters(filters) if filters else None,
            certainty=certainty,
            return_metadata=weaviate.classes.query.MetadataQuery(certainty=True, distance=True)
        )
        
        return self._format_results(response.objects)
    
    async def hybrid_search(self, query: str, alpha: float = 0.5, 
                          limit: int = 10, filters: Optional[Dict] = None):
        """Hybrid search combining vector and keyword search"""
        response = self.collection.query.hybrid(
            query=query,
            alpha=alpha,  # 0 = keyword, 1 = vector
            limit=limit,
            filters=self._build_filters(filters) if filters else None,
            fusion_type=HybridFusion.RELATIVE_SCORE,
            return_metadata=weaviate.classes.query.MetadataQuery(
                score=True,
                explain_score=True,
                certainty=True
            )
        )
        
        return self._format_results(response.objects)
    
    async def multimodal_search(self, text_query: str, image_embedding: List[float],
                              alpha: float = 0.7, limit: int = 10):
        """Multi-modal search combining text and image vectors"""
        # For multi-modal, you'd need a custom implementation
        # This is a simplified version
        response = self.collection.query.near_text(
            query=text_query,
            limit=limit,
            return_metadata=weaviate.classes.query.MetadataQuery(certainty=True)
        )
        
        return self._format_results(response.objects)
    
    async def generative_search(self, query: str, limit: int = 5,
                              generate_prompt: str = None):
        """Search with generative AI augmentation"""
        if not generate_prompt:
            generate_prompt = """
            Summarize the key points from these documents in relation to the query: {query}
            
            Documents:
            {documents}
            """
        
        response = self.collection.generate.near_text(
            query=query,
            limit=limit,
            grouped_task=generate_prompt,
            return_metadata=weaviate.classes.query.MetadataQuery(certainty=True)
        )
        
        return {
            "results": self._format_results(response.objects),
            "generated_summary": response.generated
        }
    
    async def faceted_search(self, query: str, facets: List[str],
                           limit: int = 10):
        """Search with faceted filtering and aggregation"""
        response = self.collection.aggregate.over_all(
            filters=Filter.by_property("category").equal("AI"),
            return_metrics=[weaviate.classes.query.Metrics("word_count").count().maximum().minimum().mean()]
        )
        
        search_results = await self.semantic_search(query, limit)
        
        return {
            "search_results": search_results,
            "facets": {
                "word_count_stats": response.attributes[0] if response.attributes else {}
            }
        }
    
    async def conversational_search(self, conversation_history: List[Dict],
                                  current_query: str, limit: int = 5):
        """Context-aware search using conversation history"""
        # Build context from conversation history
        context = " ".join([f"Q: {msg['query']} A: {msg.get('response', '')}" 
                          for msg in conversation_history[-3:]])  # Last 3 exchanges
        
        enhanced_query = f"Context: {context}. Current question: {current_query}"
        
        return await self.semantic_search(enhanced_query, limit)
    
    async def similarity_graph(self, doc_id: str, depth: int = 2,
                             limit_per_depth: int = 3):
        """Find similar documents and build a similarity graph"""
        similar_docs = {}
        
        # Get initial document
        initial_doc = self.collection.query.fetch_object_by_id(doc_id)
        if not initial_doc:
            return {"error": "Document not found"}
        
        similar_docs[doc_id] = {
            "document": initial_doc.properties,
            "similar": []
        }
        
        # Find similar documents recursively
        await self._find_similar_recursive(doc_id, similar_docs, depth, limit_per_depth)
        
        return similar_docs
    
    async def _find_similar_recursive(self, source_id: str, graph: Dict,
                                    depth: int, limit: int, current_depth: int = 0):
        """Recursively find similar documents"""
        if current_depth >= depth:
            return
        
        # Find similar documents
        source_doc = self.collection.query.fetch_object_by_id(source_id)
        if not source_doc:
            return
        
        similar_response = self.collection.query.near_object(
            near_object=source_id,
            limit=limit,
            return_metadata=weaviate.classes.query.MetadataQuery(certainty=True)
        )
        
        for obj in similar_response.objects:
            if obj.uuid not in graph:
                graph[obj.uuid] = {
                    "document": obj.properties,
                    "similar": []
                }
            
            # Add to similarity list
            if obj.uuid != source_id:
                graph[source_id]["similar"].append({
                    "id": obj.uuid,
                    "certainty": obj.metadata.certainty,
                    "title": obj.properties.get('title', '')
                })
                
                # Recursive call for next depth
                await self._find_similar_recursive(
                    obj.uuid, graph, depth, limit, current_depth + 1
                )
    
    def _build_filters(self, filters: Dict) -> Filter:
        """Build Weaviate filter from dictionary"""
        filter_conditions = []
        
        for field, value in filters.items():
            if isinstance(value, (list, tuple)):
                filter_conditions.append(
                    Filter.by_property(field).contains_any(value)
                )
            elif isinstance(value, str):
                filter_conditions.append(
                    Filter.by_property(field).equal(value)
                )
            elif isinstance(value, dict):
                if 'min' in value and 'max' in value:
                    filter_conditions.append(
                        Filter.by_property(field).greater_or_equal(value['min']). \
                        less_or_equal(value['max'])
                    )
        
        if len(filter_conditions) == 1:
            return filter_conditions[0]
        else:
            # Combine multiple filters with AND
            combined_filter = filter_conditions[0]
            for condition in filter_conditions[1:]:
                combined_filter = combined_filter & condition
            return combined_filter
    
    def _format_results(self, objects) -> List[Dict]:
        """Format search results for API response"""
        formatted = []
        for obj in objects:
            formatted.append({
                "id": obj.uuid,
                "title": obj.properties.get('title', ''),
                "content": obj.properties.get('content', ''),
                "category": obj.properties.get('category', ''),
                "tags": obj.properties.get('tags', []),
                "author": obj.properties.get('author', ''),
                "certainty": getattr(obj.metadata, 'certainty', None),
                "distance": getattr(obj.metadata, 'distance', None),
                "score": getattr(obj.metadata, 'score', None),
                "explanation": getattr(obj.metadata, 'explain_score', None)
            })
        return formatted

# Example usage
async def search_examples():
    from weaviate_schema import WeaviateSemanticSearch
    
    search_engine = WeaviateSemanticSearch()
    semantic_search = AdvancedSemanticSearch(search_engine.client)
    
    # Basic semantic search
    results = await semantic_search.semantic_search(
        "machine learning algorithms",
        limit=5,
        filters={"category": "AI"}
    )
    print("Semantic search results:", results)
    
    # Hybrid search
    hybrid_results = await semantic_search.hybrid_search(
        "neural networks deep learning",
        alpha=0.7,
        limit=5
    )
    print("Hybrid search results:", hybrid_results)
    
    # Generative search
    generative_results = await semantic_search.generative_search(
        "explain machine learning concepts",
        limit=3
    )
    print("Generative search results:", generative_results)

if __name__ == "__main__":
    asyncio.run(search_examples())

  

📊 Performance Optimization and Monitoring

Production vector databases require careful performance tuning and monitoring. Here are optimization strategies for 2025:

  • Index Tuning: Optimize HNSW parameters (ef, efConstruction, maxConnections) for your data
  • Sharding Strategy: Implement custom sharding based on access patterns
  • Caching Layers: Add Redis for frequent query caching
  • Batch Operations: Use batch imports and updates for better throughput
  • Query Optimization: Pre-filter when possible to reduce vector search space
  • ⚡ Key Takeaways

    1. Schema Design Matters: Proper class and property configuration significantly impacts performance and functionality
    2. Hybrid Search Excellence: Combine vector and keyword search for the best of both worlds
    3. Production Readiness: Implement proper error handling, monitoring, and backup strategies
    4. Multi-modal Capabilities: Weaviate can handle text, images, and custom embeddings simultaneously
    5. Scalability First: Design for horizontal scaling from the beginning with proper sharding
    6. Generative Integration: Leverage Weaviate's built-in generative AI modules for enhanced search
    7. Monitoring Essential: Implement comprehensive monitoring for performance and reliability

    ❓ Frequently Asked Questions

    How does Weaviate compare to other vector databases like Pinecone or Chroma?
    Weaviate stands out with its GraphQL interface, multi-modal capabilities, and built-in generative AI modules. While Pinecone excels in pure vector search performance and Chroma offers simplicity, Weaviate provides a complete ecosystem with hybrid search, filtering, and AI integration out of the box. It's particularly strong for complex applications needing both vector and traditional database features.
    What's the optimal batch size for importing data into Weaviate?
    For optimal performance, use batch sizes between 50-200 objects. Smaller batches increase network overhead, while larger batches can cause memory issues and timeouts. The sweet spot depends on your object size and network latency. Monitor your import performance and adjust accordingly. For large-scale imports, consider parallelizing across multiple workers with appropriate rate limiting.
    Can I use custom embedding models with Weaviate?
    Yes, Weaviate supports custom embedding models through its modules system. You can implement custom vectorizers or bring your own pre-computed embeddings. For custom models, you'll need to implement a vectorizer module or use the 'none' vectorizer and provide embeddings directly. This flexibility allows integration with specialized models for different domains or languages.
    How do I handle data updates and real-time synchronization?
    Weaviate supports real-time updates through its GraphQL API. For synchronization with external systems, implement change data capture (CDC) patterns or use Weaviate's webhook system. For large-scale updates, use batch operations with proper error handling. Remember that vector updates require re-embedding, so consider the computational cost of frequent updates.
    What's the best way to scale Weaviate for high-traffic applications?
    Implement horizontal scaling with proper sharding configuration, use read replicas for query load distribution, and implement caching at multiple levels. For write-heavy applications, consider sharding by time or category. Monitor performance metrics and use connection pooling. For ultimate scalability, consider Weaviate Cloud with managed scaling or Kubernetes deployment with auto-scaling.
    How do I ensure data consistency and backup in production?
    Use Weaviate's replication features for high availability, implement regular snapshot backups, and use consistent read/write consistency levels. For critical data, enable synchronous replication and regular backup exports. Monitor disk usage and implement alerting for capacity planning. Test your backup and recovery procedures regularly.

    💬 Have you implemented semantic search with Weaviate or other vector databases? Share your experiences, challenges, or performance tips in the comments below! If you found this guide helpful, please share it with your team or on social media to help others master vector databases.

    About LK-TECH Academy — Practical tutorials & explainers on software engineering, AI, and infrastructure. Follow for concise, hands-on guides.