Data Access Abstraction

Data Access Abstraction

Build APIs with swappable data layers using the repository pattern. Develop locally with SQLite, deploy to Azure with Cosmos DB or PostgreSQL — same route code, different backend. Works in any language.

When to Use This Skill

• Building an API that needs to run locally (SQLite) and in Azure (Cosmos DB or PostgreSQL)
• Adding a new data provider to an existing app without changing routes or business logic
• Migrating from one database to another incrementally
• Any project following the AIMarket journey pattern (local dev → cloud deploy)

Pattern Overview

The pattern is the same regardless of language:

Routes/Controllers → Repository Interfaces → Factory → Implementations
                                                │
                                                ├── SQLite (local dev)
                                                ├── Cosmos DB (Azure deploy)
                                                ├── PostgreSQL (Azure deploy)
                                                └── In-memory (testing)

Three rules:

1. Define repository interfaces (or abstract classes / protocols) per entity
2. Routes depend only on the interfaces — never import a database client directly
3. A factory reads a config value (DATA_PROVIDER) and returns the right implementation

Adding a new database means writing a new implementation file. Zero changes to routes.

Environment Variable

All languages use the same convention:

DATA_PROVIDER=sqlite     # Local development (default)
DATA_PROVIDER=cosmos     # Azure Cosmos DB
DATA_PROVIDER=postgres   # Azure PostgreSQL

---

Node.js / TypeScript

Repository Interfaces

// data/interfaces.ts

export interface IProductRepository {
  getAll(params: {
    page: number; pageSize: number;
    category?: string; minPrice?: number; maxPrice?: number;
  }): Promise;

  getById(id: string): Promise;
  create(input: CreateProductInput): Promise;
  update(id: string, fields: Partial): Promise;
}

export interface IOrderRepository {
  create(input: CreateOrderInput): Promise;
  getById(id: string): Promise;
  getByUserId(userId: string, page: number, pageSize: number): Promise;
}

export interface IUserRepository {
  create(input: CreateUserInput): Promise;
  getById(id: string): Promise;
  getByEmail(email: string): Promise;
}

export interface DataStore {
  products: IProductRepository;
  orders: IOrderRepository;
  users: IUserRepository;
  close?(): void;
}

Factory

// data/store.ts

export async function createStore(): Promise {
  const provider = process.env.DATA_PROVIDER || 'sqlite';
  switch (provider) {
    case 'sqlite':
      return (await import('./sqlite.js')).createSqliteStore();
    case 'cosmos':
      return (await import('./cosmos.js')).createCosmosStore();
    case 'postgres':
      return (await import('./postgres.js')).createPostgresStore();
    default:
      throw new Error(`Unknown DATA_PROVIDER: ${provider}`);
  }
}

SQLite Implementation

Use better-sqlite3 (synchronous API, fast). Wrap sync calls in Promise.resolve() or make routes await — sync values resolve immediately.

// data/sqlite.ts

import Database from 'better-sqlite3';

export function createSqliteStore(dbPath = 'app.db'): DataStore {
  const db = new Database(dbPath);
  db.pragma('journal_mode = WAL');
  db.pragma('foreign_keys = ON');
  // Create tables, seed if empty...

  return {
    products: {
      async getAll({ page, pageSize, category, minPrice, maxPrice }) {
        let where = 'WHERE status = ?';
        const params: any[] = ['active'];
        if (category) { where += ' AND category = ?'; params.push(category); }
        if (minPrice != null) { where += ' AND price >= ?'; params.push(minPrice); }
        if (maxPrice != null) { where += ' AND price  db.close(),
  };
}

SQLite-specific: Store arrays as JSON strings (JSON.stringify(tags)), parse on read. Use a junction table for order items.

Cosmos DB Implementation

// data/cosmos.ts

import { CosmosClient } from '@azure/cosmos';

export function createCosmosStore(): DataStore {
  const client = new CosmosClient({ endpoint: process.env.COSMOS_ENDPOINT!, key: process.env.COSMOS_KEY! });
  const db = client.database(process.env.COSMOS_DATABASE || 'aimarket');

  return {
    products: {
      async getAll({ page, pageSize, category, minPrice, maxPrice }) {
        let query = 'SELECT * FROM c WHERE c.status = @status';
        const parameters = [{ name: '@status', value: 'active' }];
        if (category) { query += ' AND c.category = @cat'; parameters.push({ name: '@cat', value: category }); }
        // ... build query
        const { resources } = await db.container('products').items.query({ query, parameters }).fetchAll();
        return { data: resources.slice((page-1)*pageSize, page*pageSize), totalCount: resources.length };
      },
      // ...
    },
  };
}

Cosmos-specific: Arrays and objects are native JSON — no serialization. Embed order items in the order document.

PostgreSQL Implementation

Use pg (node-postgres). PostgreSQL supports native arrays and JSONB, so it sits between SQLite and Cosmos in terms of data handling.

// data/postgres.ts

import pg from 'pg';

export function createPostgresStore(): DataStore {
  const pool = new pg.Pool({
    connectionString: process.env.POSTGRES_CONNECTION_STRING,
    ssl: process.env.POSTGRES_SSL === 'true' ? { rejectUnauthorized: false } : undefined,
  });

  return {
    products: {
      async getAll({ page, pageSize, category, minPrice, maxPrice }) {
        let where = 'WHERE status = $1';
        const params: any[] = ['active'];
        let idx = 2;

        if (category) { where += ` AND category = $${idx++}`; params.push(category); }
        if (minPrice != null) { where += ` AND price >= $${idx++}`; params.push(minPrice); }
        if (maxPrice != null) { where += ` AND price  pool.end(),
  };
}

PostgreSQL-specific: Use TEXT[] for arrays (native), JSONB for nested objects, and $1 parameterized queries. Azure PostgreSQL requires ssl: { rejectUnauthorized: false }.

Routes

// routes/products.ts — depends only on DataStore interface

export function productRoutes(store: DataStore): Router {
  const router = Router();
  router.get('/', async (req, res) => {
    const result = await store.products.getAll({ page: 1, pageSize: 20 });
    res.json(result);
  });
  return router;
}

---

Python / FastAPI

Repository Interfaces

Use Protocol (structural typing) or ABC (nominal typing). Protocol is more Pythonic.

# data/interfaces.py

from typing import Protocol
from models import Product, Order, User

class ProductRepository(Protocol):
    def get_all(self, *, page: int = 1, page_size: int = 20,
                category: str | None = None, min_price: float | None = None,
                max_price: float | None = None) -> dict:
        """Returns { 'data': list[Product], 'total_count': int }"""
        ...

    def get_by_id(self, id: str) -> Product | None: ...
    def create(self, data: dict) -> Product: ...
    def update(self, id: str, fields: dict) -> Product | None: ...

class OrderRepository(Protocol):
    def create(self, data: dict) -> Order: ...
    def get_by_id(self, id: str) -> Order | None: ...
    def get_by_user_id(self, user_id: str, page: int, page_size: int) -> dict: ...

class UserRepository(Protocol):
    def create(self, data: dict) -> User: ...
    def get_by_id(self, id: str) -> User | None: ...
    def get_by_email(self, email: str) -> User | None: ...

class DataStore(Protocol):
    products: ProductRepository
    orders: OrderRepository
    users: UserRepository

Factory

# data/store.py

import os

def create_store() -> DataStore:
    provider = os.getenv("DATA_PROVIDER", "sqlite")
    if provider == "sqlite":
        from data.sqlite import create_sqlite_store
        return create_sqlite_store()
    elif provider == "cosmos":
        from data.cosmos import create_cosmos_store
        return create_cosmos_store()
    elif provider == "postgres":
        from data.postgres import create_postgres_store
        return create_postgres_store()
    else:
        raise ValueError(f"Unknown DATA_PROVIDER: {provider}")

SQLite Implementation

# data/sqlite.py

import sqlite3, json, uuid
from models import Product

class SqliteProductRepository:
    def __init__(self, db: sqlite3.Connection):
        self.db = db

    def get_all(self, *, page=1, page_size=20, category=None, min_price=None, max_price=None):
        where, params = "WHERE status = ?", ["active"]
        if category:
            where += " AND category = ?"; params.append(category)
        if min_price is not None:
            where += " AND price >= ?"; params.append(min_price)
        if max_price is not None:
            where += " AND price  Product:
        d = dict(row)
        d["tags"] = json.loads(d["tags"])  # JSON string → list
        return Product(**d)

def create_sqlite_store(db_path="app.db"):
    db = sqlite3.connect(db_path)
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA journal_mode=WAL")
    db.execute("PRAGMA foreign_keys=ON")
    # Create tables, seed if empty...
    return SqliteDataStore(
        products=SqliteProductRepository(db),
        orders=SqliteOrderRepository(db),
        users=SqliteUserRepository(db),
    )

Cosmos DB Implementation

# data/cosmos.py

from azure.cosmos import CosmosClient
import os

class CosmosProductRepository:
    def __init__(self, container):
        self.container = container

    def get_all(self, *, page=1, page_size=20, category=None, min_price=None, max_price=None):
        query = "SELECT * FROM c WHERE c.status = @status"
        params = [{"name": "@status", "value": "active"}]
        if category:
            query += " AND c.category = @cat"
            params.append({"name": "@cat", "value": category})
        # ... build query
        items = list(self.container.query_items(query=query, parameters=params, enable_cross_partition_query=True))
        start = (page - 1) * page_size
        return {"data": items[start:start+page_size], "total_count": len(items)}

def create_cosmos_store():
    client = CosmosClient(os.environ["COSMOS_ENDPOINT"], os.environ["COSMOS_KEY"])
    db = client.get_database_client(os.getenv("COSMOS_DATABASE", "aimarket"))
    return CosmosDataStore(
        products=CosmosProductRepository(db.get_container_client("products")),
        orders=CosmosOrderRepository(db.get_container_client("orders")),
        users=CosmosUserRepository(db.get_container_client("users")),
    )

FastAPI Routes

# routes/products.py — depends only on DataStore protocol

from fastapi import APIRouter, Depends
from data.store import create_store

router = APIRouter(prefix="/api/products")

@router.get("/")
def list_products(page: int = 1, page_size: int = 20, category: str | None = None,
                  store: DataStore = Depends(create_store)):
    result = store.products.get_all(page=page, page_size=page_size, category=category)
    return {**result, "page": page, "page_size": page_size}

---

.NET / C# (Minimal APIs or Controllers)

Repository Interfaces

// Data/Interfaces.cs

public interface IProductRepository
{
    Task Data, int TotalCount)> GetAllAsync(
        int page = 1, int pageSize = 20,
        string? category = null, decimal? minPrice = null, decimal? maxPrice = null);
    Task GetByIdAsync(string id);
    Task CreateAsync(CreateProductInput input);
    Task UpdateAsync(string id, UpdateProductInput input);
}

public interface IOrderRepository
{
    Task CreateAsync(CreateOrderInput input);
    Task GetByIdAsync(string id);
    Task Data, int TotalCount)> GetByUserIdAsync(string userId, int page, int pageSize);
}

public interface IUserRepository
{
    Task CreateAsync(CreateUserInput input);
    Task GetByIdAsync(string id);
    Task GetByEmailAsync(string email);
}

Dependency Injection (Factory)

// Program.cs

var provider = builder.Configuration["DATA_PROVIDER"] ?? "sqlite";

if (provider == "sqlite")
{
    builder.Services.AddSingleton();
    builder.Services.AddSingleton();
    builder.Services.AddSingleton();
}
else if (provider == "cosmos")
{
    builder.Services.AddSingleton();
    builder.Services.AddSingleton();
    builder.Services.AddSingleton();
}
else if (provider == "postgres")
{
    builder.Services.AddSingleton();
    builder.Services.AddSingleton();
    builder.Services.AddSingleton();
}

.NET uses built-in DI instead of a manual factory. The pattern is the same: register implementations based on config, inject interfaces into routes.

SQLite Implementation

// Data/SqliteProductRepository.cs

using Microsoft.Data.Sqlite;
using System.Text.Json;

public class SqliteProductRepository : IProductRepository
{
    private readonly SqliteConnection _db;

    public SqliteProductRepository(IConfiguration config)
    {
        _db = new SqliteConnection(config.GetConnectionString("Sqlite") ?? "Data Source=app.db");
        _db.Open();
        // Create tables, seed if empty...
    }

    public async Task Data, int TotalCount)> GetAllAsync(
        int page = 1, int pageSize = 20,
        string? category = null, decimal? minPrice = null, decimal? maxPrice = null)
    {
        var where = "WHERE status = 'active'";
        if (category != null) where += $" AND category = @category";
        if (minPrice != null) where += $" AND price >= @minPrice";
        if (maxPrice != null) where += $" AND price (row["tags"])
    }
}

Cosmos DB Implementation

// Data/CosmosProductRepository.cs

using Microsoft.Azure.Cosmos;

public class CosmosProductRepository : IProductRepository
{
    private readonly Container _container;

    public CosmosProductRepository(CosmosClient client, IConfiguration config)
    {
        var db = client.GetDatabase(config["COSMOS_DATABASE"] ?? "aimarket");
        _container = db.GetContainer("products");
    }

    public async Task Data, int TotalCount)> GetAllAsync(
        int page = 1, int pageSize = 20,
        string? category = null, decimal? minPrice = null, decimal? maxPrice = null)
    {
        var query = new QueryDefinition("SELECT * FROM c WHERE c.status = @status")
            .WithParameter("@status", "active");
        // Add filters, execute, paginate...
    }
}

Minimal API Routes

// Routes depend only on interfaces — injected via DI

app.MapGet("/api/products", async (IProductRepository repo,
    int page = 1, int pageSize = 20, string? category = null) =>
{
    var (data, total) = await repo.GetAllAsync(page, pageSize, category);
    return Results.Ok(new { data, page, pageSize, totalCount = total });
});

---

Java / Spring Boot

Repository Interfaces

// data/ProductRepository.java

public interface ProductRepository {
    PaginatedResult getAll(int page, int pageSize,
        String category, Double minPrice, Double maxPrice);
    Optional getById(String id);
    Product create(CreateProductInput input);
    Optional update(String id, UpdateProductInput input);
}

// data/OrderRepository.java

public interface OrderRepository {
    Order create(CreateOrderInput input);
    Optional getById(String id);
    PaginatedResult getByUserId(String userId, int page, int pageSize);
}

// data/UserRepository.java

public interface UserRepository {
    User create(CreateUserInput input);
    Optional getById(String id);
    Optional getByEmail(String email);
}

Factory via Spring Profiles

Spring profiles replace the manual factory. Set DATA_PROVIDER as the active profile:

# application.yml
spring:
  profiles:
    active: ${DATA_PROVIDER:sqlite}

// data/sqlite/SqliteProductReposit

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [microsoft](https://github.com/microsoft)
- **Source:** [microsoft/agentic-journeys](https://github.com/microsoft/agentic-journeys)
- **License:** MIT

Install and usage instructions live in the source repository linked above.