Repository: winter-telescope/mirar (language: python)

Issue:
[FEATURE] Delete deprecated database code
**Is your feature request related to a problem? Please describe.**
We have two sets of nearly identical database code. It makes debugging/development very difficult, because it is hard to work out which version of e.g PostgresUser is doing something. 

We should just pick one and delete the other.

[FEATURE] Conventional postgres naming
**Is your feature request related to a problem? Please describe.**
I'm always frustrated when I cannot easily query mirar-built DBs in e.g dashboards, because they ignore the psql lower_case naming convention.

**Describe the solution you'd like**
Every mirar db table should follow the strong psql convention of lowercase naming for fields and tables. We could update the db exporter to automatically convert header keys to lower case for this.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/continuous_integration.yml b/.github/workflows/continuous_integration.yml
index 2edc02864..323404d4d 100644
--- a/.github/workflows/continuous_integration.yml
+++ b/.github/workflows/continuous_integration.yml
@@ -73,8 +73,13 @@ jobs:
       - name: postgres
         run: |
           sudo apt-get install postgresql
+          sudo apt-get install postgresql-common
+          sudo apt-get install postgresql-server-dev-14
           sudo service postgresql restart
           sudo -u postgres psql -c "create user runner; grant all privileges on database postgres to runner; ALTER USER runner WITH SUPERUSER;"
+          git clone https://github.com/segasai/q3c.git
+          make -C q3c
+          sudo make -C q3c install
 
       # First make sure the doc tests are up to date
       - name: Run doc tests
diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml
index 7e0b9594c..f3e6d7cb3 100644
--- a/.github/workflows/pylint.yml
+++ b/.github/workflows/pylint.yml
@@ -26,4 +26,4 @@ jobs:
           poetry install
       - name: Analysing the code with pylint
         run: |
-           poetry run pylint mirar --fail-under=9.72
+           poetry run pylint mirar --fail-under=9.78
diff --git a/docs/source/design.rst b/docs/source/design.rst
new file mode 100644
index 000000000..672e8f996
--- /dev/null
+++ b/docs/source/design.rst
@@ -0,0 +1,31 @@
+Design Philosophy
+=================
+
+mirar ia a modular python package for analaysis astronomy images.
+
+There are a few guiding principles that ensures the modularity works:
+
+- Transparent requirements: Processors should either be self-contained, or be explicit about their requirements. Processors have a dedicated function, check_prerequisites, for this purpose. The goal is that processors should work out of the box, regardless of preceding processors, or fail immediately because a required precursor is missing.
+
+- Flexibility for input images: There are a small number of fields which will always be provided in every image. These are core fields. Processors should not assume other keys are present, unless they are generated by explicitly required prerequisites.
+
+- Consistent data: astronomy data might look different when it comes out of the instrument, but when it's read in by mirar the pipeline will make sure that the data is converted to the right format. You should only ever open fits files using the open_fits function in mirar.io, and never use astropy.io directly. The open_fits is a wrapper which also performs additional checks as needed.
+
+- Consistent naming: mirar follows the official python naming conventions, as enumerated in pep:
+    - Classes are CamelCase
+    - Functions are snake_case
+    - Variables are snake_case
+    - Constants are ALL_CAPS
+    - Modules are lowercase
+    - Filenames are lowercase
+    - Directories are lowercase
+
+- Optional image saving: A lot of astronomy software will repeatedly write and rewrite science images. mirar DOES NOT DO THIS! If a processor requires a science image to be written, e.g because it is a wrapper to other software, it should be saved as a temporary file then reloaded and the temporary image deleted. The only time a science image gets permanently saved is if an ImageSaver processor is used. The same holds for masks. However, ancillary images (such as scorr images or weight image) which are generated by external tools (such as Swarp) may be written to disk by processors.
+
+- Masks as nans: mirar follows a masking convention where masked pixels are written as nabs in the science image. For conveience, a mask image can be grabbed and/or written to file, with the convewntion that 0 is masked and 1 is not masked.
+
+- Consistent logging: mirar uses the python logger. You should never include print statements. Warnings and errors should be logged at the appropriate level. Statements at the dataset level can be 'info' or 'debug', image-level statements for debugging purposes should be logged under debug.
+
+- Instrument-agnostic: Most of the code (especially processors!) is instrument-agnostic. Each instrument needs a dedicated pipeline subdirectory, containing all instrument-specific info. The rest of the code should work for any instrument.
+
+- Pythonic error handling: the code should never fail silently or return integer error codes. Instead, it should follow pythonic conventions, and explicitly raise an exception. This exception must inherit from the base ProcessorError class in mirar.errors, so it can be caught, logged and handled correctly.
diff --git a/docs/source/usage.rst b/docs/source/usage.rst
index 6d1cd4cff..aa23bc107 100644
--- a/docs/source/usage.rst
+++ b/docs/source/usage.rst
@@ -38,7 +38,7 @@ Alternatively, you can check via the code:
 .. doctest::
 
     >>> from mirar.pipelines import Pipeline
-        >>> print([x for x in Pipeline.pipelines.keys()])
-        ['sedmv2', 'summer', 'wirc', 'winter']
-    >>> print([x for x in Pipeline.pipelines.keys()])
-    ['sedmv2', 'summer', 'wirc', 'winter']
+        >>> print(sorted([x for x in Pipeline.pipelines.keys()]))
+        ['sedmv2', 'summer', 'winter', 'wirc']
+    >>> print(sorted([x for x in Pipeline.pipelines.keys()]))
+    ['sedmv2', 'summer', 'winter', 'wirc']
diff --git a/mirar/database/__init__.py b/mirar/database/__init__.py
new file mode 100644
index 000000000..9d43eab4b
--- /dev/null
+++ b/mirar/database/__init__.py
@@ -0,0 +1,3 @@
+"""
+Central module for database-related functions
+"""
diff --git a/mirar/database/base_model.py b/mirar/database/base_model.py
new file mode 100644
index 000000000..b1fbd74b1
--- /dev/null
+++ b/mirar/database/base_model.py
@@ -0,0 +1,245 @@
+"""
+Module to define PSQL database tables using sqlalchemy
+"""
+import logging
+from datetime import date
+from typing import ClassVar, Type
+
+import pandas as pd
+from psycopg import errors
+from pydantic import BaseModel, Extra, Field, root_validator, validator
+from sqlalchemy import Column, Table, inspect
+from sqlalchemy.exc import IntegrityError
+
+from mirar.database.constants import POSTGRES_DUPLICATE_PROTOCOLS
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.transactions import select_from_table
+from mirar.database.transactions.insert import _insert_in_table
+from mirar.database.transactions.update import _update_database_entry
+
+logger = logging.getLogger(__name__)
+
+
+class PydanticBase(BaseModel):
+    """
+    Base code pydantic model (no extra colunns!)
+    """
+
+    class Config:
+        """
+        Config
+        """
+
+        extra = Extra.ignore
+
+
+class BaseDB(PydanticBase):
+    """
+    Base Database Table model, requiring an associated SQLalchemy table
+    """
+
+    sql_model: ClassVar[Type[Table]]
+
+    @root_validator(pre=True)
+    def sql_model_exists_check(cls, values):
+        """
+        Validator to ensure an sql model has been specified in the child class
+        :param values: values
+        :return: values
+        """
+
+        if "sql_model" not in cls.__annotations__:
+            raise ValueError("sql model must be set in class")
+        return values
+
+    @validator("*")
+    @classmethod
+    def validate_sql(cls, value, field):
+        """
+        Validator to ensure that the field names of a pydantic model
+        match the database table
+        :param value: value
+        :param field: field
+        :return: value
+        """
+        if field.name not in cls.sql_model.__table__.columns.keys():
+            err = f"Field '{field.name}' not duplicated in {cls.sql_model}"
+            raise ValueError(err)
+        return value
+
+    def _insert_entry(
+        self,
+        returning_key_names: str | list[str] = None,
+        duplicate_protocol: str = "ignore",
+    ) -> pd.DataFrame:
+        """
+        Insert the pydantic-ified data into the corresponding sql database
+
+        :param returning_key_names: names of keys to return
+        :param duplicate_protocol: protocol to follow if duplicate entry is found
+        :return: sequence_key dataframe
+        """
+
+        assert duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS
+
+        if returning_key_names is None:
+            returning_key_names = self.get_primary_key()
+
+        if not isinstance(returning_key_names, list):
+            returning_key_names = [returning_key_names]
+
+        try:
+            res = _insert_in_table(
+                new_entry=self.dict(),
+                sql_table=self.sql_model,
+                returning_keys=returning_key_names,
+            )
+        except IntegrityError as exc:
+            if not isinstance(exc.orig, errors.UniqueViolation):
+                raise exc
+
+            db_name = self.sql_model.db_name
+
+            if duplicate_protocol == "fail":
+                err = (
+                    f"Duplicate error, entry with {self.dict()} "
+                    f"already exists in {self.sql_model.name}."
+                )
+                logger.error(err)
+                raise errors.UniqueViolation from exc
+
+            if duplicate_protocol == "replace":
+                logger.debug(f"Conflict at {exc.orig.diag.constraint_name}")
+                logger.debug(
+                    f"Found duplicate entry in {db_name} - "
+                    f"{str(exc)}."
+                    f"Replacing with a new entry."
+                )
+                self.update_entry()
+
+            else:
+                logger.debug(
+                    f"Found duplicate entry in - "
+                    f"{str(exc)}."
+                    f"Ignoring, no new entry made."
+                )
+
+            present_unique_keys = self.get_available_unique_keys()
+
+            assert len(present_unique_keys) > 0
+
+            constraints = DBQueryConstraints(
+                columns=[x.name for x in present_unique_keys],
+                accepted_values=[self.dict()[x.name] for x in present_unique_keys],
+            )
+
+            res = select_from_table(
+                sql_table=self.sql_model,
+                db_constraints=constraints,
+                output_columns=returning_key_names,
+            )
+
+        return res
+
+    def get_primary_key(self) -> str:
+        """
+        Get the primary key of the table
+
+        :return: primary key
+        """
+        primary_key = inspect(self.sql_model).primary_key[0]
+        return primary_key.name
+
+    def get_unique_keys(self) -> list[Column]:
+        """
+        Get the unique key of the table
+
+        :return: unique keys
+        """
+        cols = [x for x in self.sql_model.__table__.columns if x.unique]
+        return cols
+
+    def get_available_unique_keys(self) -> list[Column]:
+        """
+        Get the unique keys of the table which are present in the data
+
+        :return: unique keys
+        """
+        return [x for x in self.get_unique_keys() if x.name in self.dict()]
+
+    def insert_entry(
+        self, returning_key_names: str | list[str] | None = None
+    ) -> pd.DataFrame:
+        """
+        Insert the pydantic-ified data into the corresponding sql database
+
+        :param returning_key_names: names of the keys to return
+        :return: dataframe of the sequence keys
+        """
+        result = self._insert_entry(returning_key_names=returning_key_names)
+        logger.debug(f"Return result {result}")
+        return result
+
+    def _update_entry(self, update_key_names: list[str] | str | None = None):
+        """
+        Update database entry
+
+        :param update_key_names: names of keys to be updates,
+            if None, will update all keys
+        :return: None
+        """
+
+        available_unique_keys = self.get_available_unique_keys()
+
+        assert len(available_unique_keys) > 0
+
+        constraints = DBQueryConstraints(
+            columns=[x.name for x in available_unique_keys],
+            accepted_values=[self.dict()[x.name] for x in available_unique_keys],
+        )
+
+        full_dict = self.dict()
+
+        update_dict = {key: full_dict[key] for key in update_key_names}
+
+        _update_database_entry(
+            update_dict=update_dict,
+            sql_table=self.sql_model,
+            db_constraints=constraints,
+        )
+
+    def update_entry(self, update_keys=None):
+        """
+        Wrapper to update database entry. Users should override this function.
+
+        :param update_keys: keys to update
+        :return: None
+        """
+        self._update_entry(update_keys)
+
+    @classmethod
+    def _exists(cls, values, keys: str | list = None) -> bool:
+        """
+        Function to query a table and see whether an entry with key==value exists.
+        If key is None, key will default to the table's primary key.
+
+        :param values: values to query
+        :param keys: keys to query
+        :return: True if entry exists, False otherwise
+        """
+        db_constraints = DBQueryConstraints(
+            columns=keys, accepted_values=values, comparison_types="="
+        )
+
+        match = select_from_table(
+            db_constraints=db_constraints,
+            sql_table=cls.sql_model,
+        )
+        return len(match) > 0
+
+
+ra_field: float = Field(title="RA (degrees)", ge=0.0, le=360.0)
+dec_field: float = Field(title="Dec (degrees)", ge=-90.0, le=90.0)
+alt_field: float = Field(title="Alt (degrees)", ge=0.0, le=90.0)
+az_field: float = Field(title="Az (degrees)", ge=0.0, le=360.0)
+date_field: date = Field()
diff --git a/mirar/database/base_table.py b/mirar/database/base_table.py
new file mode 100644
index 000000000..f98022a79
--- /dev/null
+++ b/mirar/database/base_table.py
@@ -0,0 +1,35 @@
+"""
+Module to define PSQL database tables using sqlalchemy
+"""
+import logging
+
+from sqlalchemy import inspect
+
+logger = logging.getLogger(__name__)
+
+
+class BaseTable:
+    """
+    Parent class for database tables. Tables should inherit from this
+    and DeclarativeBase.
+    """
+
+    @property
+    def __tablename__(self):
+        raise NotImplementedError
+
+    @property
+    def db_name(self):
+        """
+        Name of the database.
+        :return: None
+        """
+        raise NotImplementedError
+
+    def get_primary_key(self) -> str:
+        """
+        Function to get primary key of table
+        Returns:
+        primary key
+        """
+        return inspect(self.__class__).primary_key[0].name
diff --git a/mirar/database/constants.py b/mirar/database/constants.py
new file mode 100644
index 000000000..c073e375f
--- /dev/null
+++ b/mirar/database/constants.py
@@ -0,0 +1,6 @@
+"""
+Central list of postgres constants
+"""
+
+
+POSTGRES_DUPLICATE_PROTOCOLS = ["fail", "ignore", "replace"]
diff --git a/mirar/database/constraints.py b/mirar/database/constraints.py
new file mode 100644
index 000000000..911448ad0
--- /dev/null
+++ b/mirar/database/constraints.py
@@ -0,0 +1,147 @@
+"""
+Module for DBQueryConstraints to carefully specify postgres query constraints
+"""
+import numpy as np
+
+POSTGRES_ACCEPTED_COMPARISONS = ["=", "<", ">", "<=", ">=", "between", "<>", "!="]
+
+
+class DBQueryConstraints:
+    """
+    Object containing one or more postgres query constraints
+    """
+
+    def __init__(
+        self,
+        columns: str | list[str] | None = None,
+        accepted_values: str
+        | int
+        | float
+        | list[str | float | int | list]
+        | None = None,
+        comparison_types: str | list[str] | None = None,
+    ):
+        self.columns = []
+        self.accepted_values = []
+        self.comparison_types = []
+
+        self.q3c_query = None
+
+        if columns is None:
+            assert accepted_values is None
+
+        else:
+            if not isinstance(columns, list):
+                columns = [columns]
+            if not isinstance(accepted_values, list):
+                accepted_values = [accepted_values]
+
+            assert len(columns) == len(accepted_values)
+
+            if comparison_types is None:
+                comparison_types = ["="] * len(accepted_values)
+            assert len(comparison_types) == len(accepted_values)
+
+            for i, column in enumerate(columns):
+                self.add_constraint(
+                    column=column,
+                    accepted_values=accepted_values[i],
+                    comparison_type=comparison_types[i],
+                )
+
+    def add_constraint(
+        self,
+        column: str,
+        accepted_values: str | int | float | tuple[float, float] | tuple[int, int],
+        comparison_type: str = "=",
+    ):
+        """
+        Add a new constraint
+
+        :param column: column
+        :param accepted_values: accepted value for comparison
+        :param comparison_type: type of comparison, e.g '='
+        :return: None
+        """
+        assert comparison_type in POSTGRES_ACCEPTED_COMPARISONS
+
+        if comparison_type == "between":
+            assert np.logical_and(
+                isinstance(accepted_values, tuple), len(accepted_values) == 2
+            )
+
+        self.columns.append(column)
+        self.accepted_values.append(accepted_values)
+        self.comparison_types.append(comparison_type)
+
+    def add_q3c_constraint(
+        self,
+        ra: float,
+        dec: float,
+        crossmatch_radius_arcsec: float,
+        ra_field_name: str = "ra",
+        dec_field_name: str = "dec",
+    ):
+        """
+        Add a q3c constraint
+
+        :param ra: ra of source
+        :param dec: dec of source
+        :param crossmatch_radius_arcsec: crossmatch radius in arcsec
+        :param ra_field_name: ra field name in database
+        :param dec_field_name: dec field name in database
+        :return: None
+        """
+
+        crossmatch_radius_deg = crossmatch_radius_arcsec / 3600.0
+
+        constraints = (
+            f"q3c_radial_query({ra_field_name},{dec_field_name},"
+            f"{ra},{dec},{crossmatch_radius_deg}) "
+        )
+
+        self.q3c_query = constraints
+
+    def __add__(self, other):
+        new = self.__class__(self.columns, self.accepted_values, self.comparison_types)
+        for args in other:
+            new.add_constraint(*args)
+        return new
+
+    def __iadd__(self, other):
+        for args in other:
+            self.add_constraint(*args)
+        return self
+
+    def __len__(self):
+        return self.columns.__len__()
+
+    def __iter__(self):
+        return iter(zip(self.columns, self.accepted_values, self.comparison_types))
+
+    def parse_constraints(
+        self,
+    ) -> str:
+        """
+        Converts the list of constraints to sql
+
+        :return: sql string
+        """
+        constraints = []
+
+        if self.q3c_query is not None:
+            constraints.append(self.q3c_query)
+
+        for i, column in enumerate(self.columns):
+            if self.comparison_types[i] == "between":
+                constraints.append(
+                    f"{column.lower()} between {self.accepted_values[i][0]} "
+                    f"and {self.accepted_values[i][1]}"
+                )
+            else:
+                constraints.append(
+                    f"{column.lower()} {self.comparison_types[i]} "
+                    f"'{self.accepted_values[i]}'"
+                )
+
+        return " AND ".join(constraints)
diff --git a/mirar/database/credentials.py b/mirar/database/credentials.py
new file mode 100644
index 000000000..17ff89a12
--- /dev/null
+++ b/mirar/database/credentials.py
@@ -0,0 +1,16 @@
+"""
+This file contains the credential keys for the database.
+"""
+import os
+
+DB_USER_KEY = "DB_USER"
+DB_PASSWORD_KEY = "DB_PWD"
+
+PG_ADMIN_USER_KEY = "PG_ADMIN_USER"
+PG_ADMIN_PWD_KEY = "PG_ADMIN_PWD"
+
+DB_USER = os.getenv(DB_USER_KEY)
+DB_PASSWORD = os.getenv(DB_PASSWORD_KEY)
+
+ADMIN_USER = os.getenv(PG_ADMIN_USER_KEY, DB_USER)
+ADMIN_PASSWORD = os.getenv(PG_ADMIN_PWD_KEY, DB_PASSWORD)
diff --git a/mirar/database/engine.py b/mirar/database/engine.py
new file mode 100644
index 000000000..14f490791
--- /dev/null
+++ b/mirar/database/engine.py
@@ -0,0 +1,27 @@
+"""
+Util functions for database interactions
+"""
+from sqlalchemy import Engine, NullPool, create_engine
+
+from mirar.database.credentials import DB_PASSWORD, DB_USER
+
+
+def get_engine(
+    db_name: str,
+    db_user: str = DB_USER,
+    db_password: str = DB_PASSWORD,
+) -> Engine:
+    """
+    Function to create a postgres engine
+
+    :param db_user: User for db
+    :param db_password: password for db
+    :param db_name: name of db
+    :return: sqlalchemy engine
+    """
+
+    return create_engine(
+        f"postgresql+psycopg://{db_user}:{db_password}" f"@/{db_name}",
+        future=True,
+        poolclass=NullPool,
+    )
diff --git a/mirar/database/errors.py b/mirar/database/errors.py
new file mode 100644
index 000000000..958c71527
--- /dev/null
+++ b/mirar/database/errors.py
@@ -0,0 +1,8 @@
+"""
+Database errors
+"""
+from mirar.errors.exceptions import ProcessorError
+
+
+class DataBaseError(ProcessorError):
+    """Error relating to postgres interactions"""
diff --git a/mirar/database/q3c.py b/mirar/database/q3c.py
new file mode 100644
index 000000000..d1f131618
--- /dev/null
+++ b/mirar/database/q3c.py
@@ -0,0 +1,42 @@
+"""
+Module to create q3c extension and index on table
+"""
+import logging
+
+from sqlalchemy import DDL
+
+from mirar.database.engine import get_engine
+
+logger = logging.getLogger(__name__)
+
+
+def create_q3c_extension(
+    db_name: str,
+    table_name: str,
+    ra_column_name: str,
+    dec_column_name: str,
+):
+    """
+    Function to create q3c extension and index on table
+
+    :param db_name: Name of database
+    :param table_name: Name of table
+    :param ra_column_name: ra column name
+    :param dec_column_name: dec column name
+    :return:
+    """
+
+    logger.info(f"Creating q3c extension and index on table {table_name}...")
+
+    trig_ddl = DDL(
+        f"CREATE INDEX ON {table_name} "
+        f"(q3c_ang2ipix({ra_column_name}, {dec_column_name}));"
+        f"CLUSTER {table_name} USING {table_name}_q3c_ang2ipix_idx;"
+        f"ANALYZE {table_name};"
+    )
+
+    engine = get_engine(db_name=db_name)
+
+    with engine.connect() as conn:
+        conn.execute(trig_ddl)
+        conn.commit()
diff --git a/mirar/database/setup.py b/mirar/database/setup.py
new file mode 100644
index 000000000..e1e5d5c09
--- /dev/null
+++ b/mirar/database/setup.py
@@ -0,0 +1,54 @@
+"""
+Central function to setup database
+"""
+import logging
+from typing import Union
+
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import DeclarativeBase
+
+from mirar.database.base_table import BaseTable
+from mirar.database.credentials import DB_PASSWORD, DB_USER
+from mirar.database.engine import get_engine
+from mirar.database.user import PostgresAdmin, PostgresUser
+
+logger = logging.getLogger(__name__)
+
+
+def setup_database(db_base: Union[DeclarativeBase, BaseTable]):
+    """
+    Function to setup database
+
+    :param db_base: BaseTable
+    :return: None
+    """
+    db_name = db_base.db_name
+    engine = get_engine(db_name=db_name)
+
+    pg_user = PostgresUser()
+
+    try:
+        pg_user.validate_credentials()
+    except OperationalError:
+        logger.warning(
+            f"Failed to credentials for user {DB_USER}. "
+            f"Will try creating new user with this name using admin credentials."
+        )
+        pg_admin = PostgresAdmin()
+        pg_admin.validate_credentials()
+        pg_admin.create_new_user(new_db_user=DB_USER, new_password=DB_PASSWORD)
+        pg_user.validate_credentials()
+
+    pg_user.create_db(db_name=db_name)
+
+    db_base.metadata.create_all(engine)
+
+    if not pg_user.has_extension(extension_name="q3c", db_name=db_name):
+        logger.info(
+            f"No 'q3c' extension found. Creating it now for db "
+            f"{db_name} with admin user."
+        )
+
+        pg_admin = PostgresAdmin()
+        pg_admin.validate_credentials()
+        pg_admin.create_extension(extension_name="q3c", db_name=db_name)
diff --git a/mirar/database/transactions/__init__.py b/mirar/database/transactions/__init__.py
new file mode 100644
index 000000000..ce957638d
--- /dev/null
+++ b/mirar/database/transactions/__init__.py
@@ -0,0 +1,10 @@
+"""
+Central module for all DB transaction types.
+"""
+from mirar.database.transactions.insert import _insert_in_table
+from mirar.database.transactions.select import (
+    check_table_exists,
+    is_populated,
+    select_from_table,
+)
+from mirar.database.transactions.update import _update_database_entry
diff --git a/mirar/database/transactions/insert.py b/mirar/database/transactions/insert.py
new file mode 100644
index 000000000..03d14344f
--- /dev/null
+++ b/mirar/database/transactions/insert.py
@@ -0,0 +1,52 @@
+"""
+Module to export data to a database table
+"""
+
+import logging
+from typing import Type
+
+import pandas as pd
+from sqlalchemy import Insert, column
+
+from mirar.database.base_table import BaseTable
+from mirar.database.constants import POSTGRES_DUPLICATE_PROTOCOLS
+from mirar.database.engine import get_engine
+
+logger = logging.getLogger(__name__)
+
+
+def _insert_in_table(
+    new_entry: dict,
+    sql_table: Type[BaseTable],
+    duplicate_protocol: str = "fail",
+    returning_keys: list[str] | str = None,
+) -> pd.DataFrame:
+    """
+    Export a list of fields in value dict to a database table
+
+    :param new_entry: dictionary/DataBlock/other dictonary-like object to export
+    :param sql_table: table of DB to export to
+    :param returning_keys: keys to return (default: all)
+    :return:
+    """
+
+    assert duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS
+
+    if not isinstance(returning_keys, list):
+        returning_keys = [returning_keys]
+
+    db_name = sql_table.db_name
+
+    stmt = (
+        Insert(sql_table)
+        .values(new_entry)
+        .returning(*[column(x) for x in returning_keys])
+    )
+
+    engine = get_engine(db_name=db_name)
+
+    with engine.connect() as conn:
+        res = conn.execute(stmt)
+        conn.commit()
+
+    return pd.DataFrame(res.fetchall())
diff --git a/mirar/database/transactions/select.py b/mirar/database/transactions/select.py
new file mode 100644
index 000000000..8643a8e47
--- /dev/null
+++ b/mirar/database/transactions/select.py
@@ -0,0 +1,111 @@
+"""
+Module to select database entries
+"""
+import pandas as pd
+from sqlalchemy import Select, text
+
+from mirar.database.base_table import BaseTable
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.engine import get_engine
+
+
+def run_select(
+    query: Select,
+    sql_table: BaseTable,
+    columns: list[str] | None = None,
+) -> pd.DataFrame:
+    """
+    Run a select query
+
+    :param query: select query to run
+    :param sql_table: table to run query on
+    :param columns: columns to output (default: all)
+    :return: results
+    """
+
+    engine = get_engine(db_name=sql_table.db_name)
+
+    with engine.connect() as conn:
+        res = pd.read_sql(query, conn, columns=columns)
+
+    if (columns is not None) & len(res) > 0:
+        res = res[columns]
+
+    return res
+
+
+def select_from_table(
+    db_constraints: DBQueryConstraints,
+    sql_table: BaseTable,
+    output_columns: list[str] | None = None,
+) -> pd.DataFrame:
+    """
+    Select database entries
+
+    :param db_constraints: database query constraints
+    :param sql_table: database SQL table
+    :param output_columns: columns to output (default: all)
+    :return: results
+    """
+
+    res = run_select(
+        query=Select(sql_table).where(text(db_constraints.parse_constraints())),
+        sql_table=sql_table,
+        columns=output_columns,
+    )
+
+    return res
+
+
+def check_table_exists(
+    sql_table: BaseTable,
+) -> bool:
+    """
+    Check if a table exists
+
+    :param sql_table: database SQL table
+    :return: True if table exists, False otherwise
+    """
+
+    engine = get_engine(db_name=sql_table.db_name)
+
+    with engine.connect() as conn:
+        res = conn.execute(
+            text(
+                f"""
+                SELECT EXISTS (
+                    SELECT FROM information_schema.tables
+                    WHERE  table_schema = 'public'
+                    AND    table_name   = '{sql_table.__tablename__}'
+                );
+                """
+            )
+        )
+
+    return res.fetchone()[0]
+
+
+def is_populated(
+    sql_table: BaseTable,
+) -> bool:
+    """
+    Function to check if a table is populated
+
+    :param sql_table: database SQL table
+    :return: boolean
+    """
+
+    engine = get_engine(db_name=sql_table.db_name)
+
+    with engine.connect() as conn:
+        res = conn.execute(
+            text(
+                f"""
+                SELECT EXISTS (
+                    SELECT FROM {sql_table.__tablename__}
+                );
+                """
+            )
+        )
+
+    return res.fetchone()[0]
diff --git a/mirar/database/transactions/update.py b/mirar/database/transactions/update.py
new file mode 100644
index 000000000..8cd29f77a
--- /dev/null
+++ b/mirar/database/transactions/update.py
@@ -0,0 +1,35 @@
+"""
+Module to update database entries
+"""
+from sqlalchemy import text, update
+
+from mirar.database.base_table import BaseTable
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.engine import get_engine
+
+
+def _update_database_entry(
+    update_dict: dict,
+    db_constraints: DBQueryConstraints,
+    sql_table: BaseTable,
+):
+    """
+    Update a database entry
+
+    :param update_dict: dictionary of values to update
+    :param db_constraints: database query constraints
+    :param sql_table: database SQL table
+    :return: None
+    """
+
+    engine = get_engine(db_name=sql_table.db_name)
+
+    with engine.connect() as conn:
+        stmt = (
+            update(sql_table)
+            .where(text(db_constraints.parse_constraints()))
+            .values(**update_dict)
+        )
+
+        conn.execute(stmt)
+        conn.commit()
diff --git a/mirar/database/user/__init__.py b/mirar/database/user/__init__.py
new file mode 100644
index 000000000..186ac7c32
--- /dev/null
+++ b/mirar/database/user/__init__.py
@@ -0,0 +1,5 @@
+"""
+Module containing classes for interacting with postgres database
+"""
+from mirar.database.user.postgres_admin import PostgresAdmin
+from mirar.database.user.postgres_user import PostgresUser
diff --git a/mirar/database/user/postgres_admin.py b/mirar/database/user/postgres_admin.py
new file mode 100644
index 000000000..5d62ed360
--- /dev/null
+++ b/mirar/database/user/postgres_admin.py
@@ -0,0 +1,65 @@
+"""
+Postgres Admin class
+"""
+from sqlalchemy.sql.ddl import DDL
+
+from mirar.database.credentials import (
+    ADMIN_PASSWORD,
+    ADMIN_USER,
+    PG_ADMIN_PWD_KEY,
+    PG_ADMIN_USER_KEY,
+)
+from mirar.database.engine import get_engine
+from mirar.database.user.postgres_user import PostgresUser
+
+
+class PostgresAdmin(PostgresUser):
+    """
+    An Admin postgres user, with additional functionality for creatying new users
+    """
+
+    user_env_varaiable = PG_ADMIN_USER_KEY
+    pass_env_variable = PG_ADMIN_PWD_KEY
+
+    def __init__(self, db_user: str = ADMIN_USER, db_password: str = ADMIN_PASSWORD):
+        super().__init__(db_user=db_user, db_password=db_password)
+
+    def create_new_user(self, new_db_user: str, new_password: str):
+        """
+        Create a new postgres user
+
+        :param new_db_user: new username
+        :param new_password: new user password
+        :return: None
+        """
+        engine = get_engine(
+            db_name="postgres",
+            db_user=self.db_user,
+            db_password=self.db_password,
+        )
+        with engine.connect() as conn:
+            command = DDL(
+                f"CREATE ROLE {new_db_user} WITH password '{new_password}' CREATEDB NOCREATEROLE LOGIN;"
+            )
+            conn.execute(command)
+            conn.commit()
+
+    def create_extension(self, extension_name: str, db_name: str):
+        """
+        Function to create new extension for database
+
+        :param extension_name: name of extension to create
+        :param db_name: name of database to create extension in
+        :return: None
+        """
+        engine = get_engine(
+            db_name=db_name,
+            db_user=self.db_user,
+            db_password=self.db_password,
+        )
+        with engine.connect() as conn:
+            command = DDL(f"CREATE EXTENSION IF NOT EXISTS {extension_name};")
+            conn.execute(command)
+            conn.commit()
+
+        assert self.has_extension(extension_name=extension_name, db_name=db_name)
diff --git a/mirar/database/user/postgres_user.py b/mirar/database/user/postgres_user.py
new file mode 100644
index 000000000..11a765e2c
--- /dev/null
+++ b/mirar/database/user/postgres_user.py
@@ -0,0 +1,101 @@
+"""
+Module containing postgres util functions
+"""
+import logging
+
+from sqlalchemy import text
+from sqlalchemy_utils import create_database, database_exists
+
+from mirar.database.credentials import (
+    DB_PASSWORD,
+    DB_PASSWORD_KEY,
+    DB_USER,
+    DB_USER_KEY,
+)
+from mirar.database.engine import get_engine
+from mirar.database.errors import DataBaseError
+
+logger = logging.getLogger(__name__)
+
+
+class PostgresUser:
+    """
+    Basic Postgres user class for executing functions
+    """
+
+    user_env_varaiable = DB_USER_KEY
+    pass_env_variable = DB_PASSWORD_KEY
+
+    def __init__(self, db_user: str = DB_USER, db_password: str = DB_PASSWORD):
+        self.db_user = db_user
+        self.db_password = db_password
+
+    def validate_credentials(self):
+        """
+        Checks that user credentials exist
+
+        :return: None
+        """
+        if self.db_user is None:
+            err = (
+                f"'db_user' is set as None. Please pass a db_user as an argument, "
+                f"or set the environment variable '{self.user_env_varaiable}'."
+            )
+            logger.error(err)
+            raise DataBaseError(err)
+
+        if self.db_password is None:
+            err = (
+                f"'db_password' is set as None. Please pass a password as an argument, "
+                f"or set the environment variable '{self.pass_env_variable}'."
+            )
+            logger.error(err)
+            raise DataBaseError(err)
+
+        engine = get_engine(
+            db_name="postgres",
+            db_user=self.db_user,
+            db_password=self.db_password,
+        )
+
+        with engine.connect() as conn:
+            conn.execute(text("SELECT 1"))
+
+    @staticmethod
+    def create_db(db_name: str):
+        """
+        Creates a database using credentials, if it does not exist
+
+        :param db_name: DB to create
+        :return: None
+        """
+
+        engine = get_engine(db_name=db_name)
+        if not database_exists(engine.url):
+            create_database(engine.url)
+
+        assert database_exists(engine.url)
+
+    @staticmethod
+    def has_extension(
+        extension_name: str,
+        db_name: str,
+    ) -> bool:
+        """
+        Function to create q3c extension and index on table
+
+        :param extension_name: name of extension to check
+        :param db_name: name of database to check
+        :return: boolean extension exists
+        """
+
+        engine = get_engine(db_name=db_name)
+        with engine.connect() as conn:
+            command = text(
+                f"SELECT extname FROM pg_extension WHERE extname='{extension_name}'"
+            )
+            res = conn.execute(command).all()
+
+        assert len(res) <= 1, "More than one extension found"
+
+        return len(res) == 1
diff --git a/mirar/database/utils.py b/mirar/database/utils.py
new file mode 100644
index 000000000..b5493f9b1
--- /dev/null
+++ b/mirar/database/utils.py
@@ -0,0 +1,36 @@
+"""
+Module for postgres utilities
+"""
+import logging
+
+import numpy as np
+from sqlalchemy import text
+
+from mirar.database.engine import get_engine
+
+logger = logging.getLogger(__name__)
+
+
+def get_sequence_key_names_from_table(
+    db_table: str,
+    db_name: str,
+) -> list:
+    """
+    Gets sequence keys of db table
+
+    :param db_table: database table to use
+    :param db_name: database name
+    :return: numpy array of keys
+    """
+    engine = get_engine(db_name=db_name)
+    with engine.connect() as conn:
+        res = conn.execute(
+            text("SELECT c.relname FROM pg_class c WHERE c.relkind = 'S';")
+        ).fetchall()
+
+    sequences = [x[0] for x in res]
+    seq_tables = np.array([x.split("_")[0] for x in sequences])
+    seq_columns = np.array([x.split("_")[1] for x in sequences])
+    table_sequence_keys = seq_columns[(seq_tables == db_table)]
+
+    return list(table_sequence_keys)
diff --git a/mirar/paths.py b/mirar/paths.py
index 21007a7c5..efb2278a3 100644
--- a/mirar/paths.py
+++ b/mirar/paths.py
@@ -274,9 +274,6 @@ def get_astrometry_keys() -> list:
 MAGERR_PSF_KEY = "sigmapsf"
 XPOS_KEY = "xpos"
 YPOS_KEY = "ypos"
-CAND_NAME_KEY = "objectId"
-CAND_RA_KEY = "ra"
-CAND_DEC_KEY = "dec"
 FITS_MASK_KEY = "MASKFITS"
 sextractor_checkimg_keys = {
     "BACKGROUND": "BKGPT",
@@ -308,6 +305,11 @@ def get_astrometry_keys() -> list:
     BASE_NAME_KEY,
 ]
 
+CAND_NAME_KEY = "objectid"
+CAND_RA_KEY = "ra"
+CAND_DEC_KEY = "dec"
+SOURCE_HISTORY_KEY = "prv_candidates"
+
 MONITOR_EMAIL_KEY = "WATCHDOG_EMAIL"
 MONITOR_PASSWORD_KEY = "WATCHDOG_EMAIL_PASSWORD"
 MONITOR_RECIPIENT_KEY = "WATCHDOG_EMAIL_RECIPIENTS"
diff --git a/mirar/pipelines/summer/blocks.py b/mirar/pipelines/summer/blocks.py
index a15d701b1..c6c1f1b0a 100644
--- a/mirar/pipelines/summer/blocks.py
+++ b/mirar/pipelines/summer/blocks.py
@@ -4,13 +4,11 @@
 lists which are used to build configurations for the
 :class:`~mirar.pipelines.summer.summer_pipeline.SummerPipeline`.
 """
+# pylint: disable=duplicate-code
 from mirar.downloader.get_test_data import get_test_data_dir
 from mirar.paths import BASE_NAME_KEY, GAIN_KEY, OBSCLASS_KEY, core_fields
 from mirar.pipelines.summer.config import (
-    DB_NAME,
-    PIPELINE_NAME,
     SUMMER_PIXEL_SCALE,
-    get_summer_schema_path,
     psfex_config_path,
     scamp_path,
     sextractor_astrometry_config,
@@ -34,16 +32,14 @@
     load_proc_summer_image,
     load_raw_summer_image,
 )
-from mirar.pipelines.summer.models import Exposure, Proc, Raw
+from mirar.pipelines.summer.models import Diff, Exposure, Proc, Raw
 from mirar.processors import BiasCalibrator, FlatCalibrator
 from mirar.processors.astromatic import PSFex, Scamp, Sextractor, Swarp
 from mirar.processors.astrometry.autoastrometry import AutoAstrometry
 from mirar.processors.cosmic_rays import LACosmicCleaner
 from mirar.processors.csvlog import CSVLog
-from mirar.processors.database.database_exporter import (
-    DatabaseImageExporter as PSQLDatabaseImageExporter,
-)
-from mirar.processors.database.database_modifier import ModifyImageDatabaseSeq
+from mirar.processors.database.database_inserter import DatabaseImageInserter
+from mirar.processors.database.database_updater import ImageSequenceDatabaseUpdater
 from mirar.processors.mask import MaskPixelsFromPath
 from mirar.processors.photcal import PhotCalibrator
 from mirar.processors.photometry.aperture_photometry import CandidateAperturePhotometry
@@ -52,7 +48,6 @@
 from mirar.processors.sources import SourceWriter
 from mirar.processors.sources.source_detector import SourceDetector
 from mirar.processors.sources.utils import RegionsWriter
-from mirar.processors.sqldatabase.database_exporter import DatabaseImageExporter
 from mirar.processors.utils import ImageBatcher, ImageLoader, ImageSaver, ImageSelector
 from mirar.processors.utils.cal_hunter import CalHunter
 from mirar.processors.utils.header_annotate import HeaderEditor
@@ -117,13 +112,12 @@
 
 export_raw = [
     # ImageSelector(("BASENAME", "SUMMER_20220402_214324_Camera0.fits")),
-    DatabaseImageExporter(
+    DatabaseImageInserter(
         db_table=Exposure,
         duplicate_protocol="replace",
-        q3c_bool=False,
     ),
     MaskPixelsFromPath(mask_path=summer_mask_path),
-    DatabaseImageExporter(db_table=Raw, duplicate_protocol="replace", q3c_bool=False),
+    DatabaseImageInserter(db_table=Raw, duplicate_protocol="replace"),
     ImageSelector((OBSCLASS_KEY, ["bias", "flat", "science"])),
 ]
 
@@ -188,18 +182,11 @@
         # additional_headers=["PROCIMG"],
         write_mask=True,
     ),
-    HeaderEditor(edit_keys="procflag", values=1),
-    # PSQLDatabaseImageExporter(
-    #     db_name=DB_NAME,
-    #     db_table="proc",  # FIXME
-    #     schema_path=get_summer_schema_path("proc"),
-    #     duplicate_protocol="replace",
-    # ),
-    DatabaseImageExporter(db_table=Proc, duplicate_protocol="replace", q3c_bool=False),
-    ModifyImageDatabaseSeq(
-        db_name=DB_NAME,
-        db_table="raw",  # FIXME
-        schema_path=get_summer_schema_path("raw"),
+    HeaderEditor(edit_keys="procstatus", values=1),
+    DatabaseImageInserter(db_table=Proc, duplicate_protocol="replace"),
+    ImageSequenceDatabaseUpdater(
+        # db_name=DB_NAME,
+        db_table=Raw,
         db_alter_columns="procstatus",
     ),
 ]
@@ -234,10 +221,8 @@
 ]
 
 export_diff_to_db = [
-    PSQLDatabaseImageExporter(
-        db_name=PIPELINE_NAME,  # FIXME
-        db_table="diff",
-        schema_path=get_summer_schema_path("diff"),
+    DatabaseImageInserter(
+        db_table=Diff,
     ),
 ]
 
diff --git a/mirar/pipelines/summer/config/__init__.py b/mirar/pipelines/summer/config/__init__.py
index 6c8bcfc94..d2b860afc 100644
--- a/mirar/pipelines/summer/config/__init__.py
+++ b/mirar/pipelines/summer/config/__init__.py
@@ -9,7 +9,6 @@
     SUMMER_GAIN,
     SUMMER_PIXEL_SCALE,
 )
-from mirar.pipelines.summer.config.schema import get_summer_schema_path
 from mirar.processors.utils.cal_hunter import CalRequirement
 
 summer_dir = os.path.dirname(__file__)
diff --git a/mirar/pipelines/summer/config/schema/__init__.py b/mirar/pipelines/summer/config/schema/__init__.py
deleted file mode 100644
index 6f296e8ce..000000000
--- a/mirar/pipelines/summer/config/schema/__init__.py
+++ /dev/null
@@ -1,10 +0,0 @@
-"""
-Module for summer-specific db creation with psycopg
-"""
-import os
-
-summer_schema_dir = os.path.dirname(__file__)
-
-
-def get_summer_schema_path(db_name: str) -> str:
-    return os.path.join(summer_schema_dir, f"{db_name}.sql")
diff --git a/mirar/pipelines/summer/config/schema/diff.sql b/mirar/pipelines/summer/config/schema/diff.sql
deleted file mode 100644
index 86bcfecd2..000000000
--- a/mirar/pipelines/summer/config/schema/diff.sql
+++ /dev/null
@@ -1,30 +0,0 @@
-CREATE TABLE IF NOT EXISTS diff (
-    diffid BIGINT PRIMARY KEY,
-    rawid BIGINT,
-    uexpid BIGINT,
-    rawpath VARCHAR(255),
-    savepath VARCHAR(255),
-    DIFFPSF VARCHAR(255),
-    DIFFSCR VARCHAR(255),
-    DIFFUNC VARCHAR(255),
-    OBSDATE INT,
-    timeutc TIMESTAMPTZ,
-    FILTER VARCHAR(10),
-    EXPTIME REAL,
-    NIGHT INT,
-    fieldID INT,
-    CENTRA REAL,
-    CENTDEC REAL,
-    ZP_AUTO REAL,
-    SCORSTD REAL,
-    SCORMED REAL,
-    SCORMEAN REAL,
-    DIFFMLIM REAL,
-    diffcount SERIAL,
-    CONSTRAINT fk_expid
-            FOREIGN KEY(uexpid)
-                REFERENCES exposures(uexpid),
-    CONSTRAINT fk_procid
-            FOREIGN KEY(rawid)
-                REFERENCES proc(rawid)
-);
diff --git a/mirar/pipelines/summer/config/schema/proc.sql b/mirar/pipelines/summer/config/schema/proc.sql
deleted file mode 100644
index 5290824da..000000000
--- a/mirar/pipelines/summer/config/schema/proc.sql
+++ /dev/null
@@ -1,35 +0,0 @@
-CREATE TABLE IF NOT EXISTS proc (
-    procid BIGINT PRIMARY KEY,
-    expid BIGINT,
-    rawpath VARCHAR(255),
-    savepath VARCHAR(255),
-    OBSDATE INT,
-    timeutc TIMESTAMPTZ,
-    FILTER VARCHAR(10),
-    EXPTIME REAL,
-    fieldID VARCHAR(30),
-    CD1_1 REAL,
-    CD1_2 REAL,
-    CD2_1 REAL,
-    CD2_2 REAL,
-    CRVAL1 REAL,
-    CRVAL2 REAL,
-    CRPIX1 REAL,
-    CRPIX2 REAL,
-    ZP_AUTO REAL,
-    FWHM_MED REAL,
-    FWHM_STD REAL,
-    flatname VARCHAR(255),
-    biasname VARCHAR(255),
-    ASTR_DPA FLOAT,
-    ASTR_OFF FLOAT,
-    ZP_AUTO_NSTARS INT,
-    ZP_AUTO_STD FLOAT,
-    OBSMJD FLOAT,
-    progid INT,
-    proccount SERIAL,
-    MAGLIM REAL,
-    CONSTRAINT fk_expid
-            FOREIGN KEY(expid)
-                REFERENCES raw(expid)
-);
diff --git a/mirar/pipelines/summer/config/schema/raw.sql b/mirar/pipelines/summer/config/schema/raw.sql
deleted file mode 100644
index ccb04e6b6..000000000
--- a/mirar/pipelines/summer/config/schema/raw.sql
+++ /dev/null
@@ -1,40 +0,0 @@
-CREATE TABLE IF NOT EXISTS raw (
-    expid BIGINT PRIMARY KEY,
-    savepath VARCHAR(255),
-    obsdate INT,
-    timeutc TIMESTAMPTZ,
-    obsID INT,
-    itid INT,
-    night INT,
-    fieldID INT,
-    filter VARCHAR(5),
-    progID INT,
-    AExpTime FLOAT,
-    expMJD FLOAT,
-    subprog VARCHAR(20),
-    airmass FLOAT,
-    shutopen REAL,
-    shutclsd REAL,
-    tempture REAL,
-    windspd REAL,
-    Dewpoint REAL,
-    Humidity REAL,
-    Pressure REAL,
-    Moonra REAL,
-    Moondec REAL,
-    Moonillf REAL,
-    Moonphas REAL,
-    Moonalt REAL,
-    Sunaz REAL,
-    Sunalt REAL,
-    Detsoft VARCHAR(50),
-    Detfirm VARCHAR(50),
-    ra REAL,
-    dec REAL,
-    altitude REAL,
-    azimuth REAL,
-    procflag INT,
-    rawcount SERIAL
-);
-
-CREATE INDEX raw_obsdate_idx ON raw (obsdate DESC);
diff --git a/mirar/pipelines/summer/models/__init__.py b/mirar/pipelines/summer/models/__init__.py
index 1cf15c3d5..24eff86c8 100644
--- a/mirar/pipelines/summer/models/__init__.py
+++ b/mirar/pipelines/summer/models/__init__.py
@@ -1,8 +1,10 @@
 """
 Models for database and pydantic dataclass models
 """
-from typing import Type
-
+# pylint: disable=duplicate-code
+from mirar.database.credentials import DB_USER
+from mirar.database.setup import setup_database
+from mirar.pipelines.summer.models._diff import Diff, DiffTable
 from mirar.pipelines.summer.models._exposures import Exposure, ExposuresTable
 from mirar.pipelines.summer.models._fields import (
     DEFAULT_FIELD,
@@ -39,44 +41,6 @@
     populate_subdets,
 )
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseTable
-from mirar.processors.sqldatabase.postgres import PostgresAdmin
-from mirar.processors.sqldatabase.postgres_utils import (
-    ADMIN_PASSWORD,
-    ADMIN_USER,
-    DB_PASSWORD,
-    DB_USER,
-)
-from mirar.utils.sql import get_engine
-
-
-def setup_database(base: Type[BaseTable]):
-    """
-    Function to setup database
-    Args:
-        base:
-    Returns:
-    """
-    if DB_USER is not None:
-        db_name = str(base.db_name)
-        admin_engine = get_engine(
-            db_name=db_name, db_user=ADMIN_USER, db_password=ADMIN_PASSWORD
-        )
-
-        pg_admin = PostgresAdmin()
-
-        if not pg_admin.check_if_db_exists(db_name=db_name):
-            pg_admin.create_db(db_name=db_name)
-
-        base.metadata.create_all(
-            admin_engine
-        )  # extensions need to be created as a superuser
-
-        if not pg_admin.check_if_user_exists(user_name=DB_USER):
-            pg_admin.create_new_user(new_db_user=DB_USER, new_password=DB_PASSWORD)
-
-        pg_admin.grant_privileges(db_name=db_name, db_user=DB_USER)
-
 
 if DB_USER is not None:
     setup_database(SummerBase)
diff --git a/mirar/pipelines/summer/models/_diff.py b/mirar/pipelines/summer/models/_diff.py
new file mode 100644
index 000000000..52fa74c6e
--- /dev/null
+++ b/mirar/pipelines/summer/models/_diff.py
@@ -0,0 +1,77 @@
+"""
+Models for the 'diff' table
+"""
+# pylint: disable=duplicate-code
+import os
+from typing import ClassVar
+
+from pydantic import Field, validator
+from sqlalchemy import VARCHAR, Column, Double, ForeignKey, Integer, Sequence
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from mirar.database.base_model import BaseDB
+from mirar.pipelines.summer.models._exposures import Exposure
+from mirar.pipelines.summer.models.base_model import SummerBase
+
+
+class DiffTable(SummerBase):  # pylint: disable=too-few-public-methods
+    """
+    Diff table in database
+    """
+
+    __tablename__ = "diff"
+    __table_args__ = {"extend_existing": True}
+
+    udiffid = Column(
+        Integer,
+        Sequence(start=1, name="diff_udiffid_seq"),
+        autoincrement=True,
+        unique=True,
+    )
+    diffid = Column(Double, primary_key=True, autoincrement=False)
+    proc: Mapped["ProcTable"] = relationship(back_populates="diff_ids")
+
+    uexpid: Mapped[int] = mapped_column(ForeignKey("exposures.uexpid"))
+    exposure_ids: Mapped["ExposuresTable"] = relationship(back_populates="diff")
+
+    savepath = Column(VARCHAR(255), unique=True)
+
+
+class Diff(BaseDB):
+    """
+    A pydantic model for a diff database entry
+    """
+
+    sql_model: ClassVar = DiffTable
+
+    diffid: int = Field(ge=0)
+    uexpid: int = Field(ge=0)
+    qid: int = Field(ge=0)
+    savepath: str = Field(min_length=1)
+    procstatus: int = Field(ge=0, default=0)
+
+    @validator("savepath")
+    @classmethod
+    def validate_savepath(cls, field_value: str):
+        """
+        Ensure that path exists
+
+        :param field_value: field value
+        :return: field value
+        """
+        assert os.path.exists(field_value)
+        return field_value
+
+    @validator("uexpid")
+    @classmethod
+    def validate_expid(cls, field_value: int):
+        """
+        Ensure that expid exists in exposures table
+        Args:
+            field_value: expid
+
+        Returns:
+
+        """
+        assert Exposure._exists(keys="uexpid", values=field_value)
+        return field_value
diff --git a/mirar/pipelines/summer/models/_exposures.py b/mirar/pipelines/summer/models/_exposures.py
index adbf3f073..27723aabc 100644
--- a/mirar/pipelines/summer/models/_exposures.py
+++ b/mirar/pipelines/summer/models/_exposures.py
@@ -1,17 +1,18 @@
 """
 Models for the 'exposures' table
 """
-# from mirar.utils.sql import create_q3c_extension
+# pylint: disable=duplicate-code
 import logging
 from datetime import date, datetime
 from typing import ClassVar
 
+import pandas as pd
 from pydantic import Field
-from sqlalchemy import (  # event,
+from sqlalchemy import (
     VARCHAR,
+    BigInteger,
     Column,
     DateTime,
-    Double,
     Float,
     ForeignKey,
     Integer,
@@ -19,19 +20,13 @@
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+from mirar.database.base_model import BaseDB, alt_field, az_field, dec_field, ra_field
 from mirar.pipelines.summer.models._fields import FieldsTable, fieldid_field
 from mirar.pipelines.summer.models._filters import FiltersTable, fid_field
 from mirar.pipelines.summer.models._img_type import ImgTypesTable
 from mirar.pipelines.summer.models._nights import Night, NightsTable
-from mirar.pipelines.summer.models._programs import ProgramsTable, default_program
+from mirar.pipelines.summer.models._programs import Program
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import (
-    BaseDB,
-    alt_field,
-    az_field,
-    dec_field,
-    ra_field,
-)
 
 logger = logging.getLogger(__name__)
 
@@ -49,8 +44,9 @@ class ExposuresTable(SummerBase):  # pylint: disable=too-few-public-methods
         Sequence(name="exposures_uexpid_seq", start=1, increment=1),
         unique=True,
         autoincrement=True,
+        primary_key=True,
     )
-    expid = Column(Double, primary_key=True, unique=True, autoincrement=False)
+    expid = Column(BigInteger, unique=True, autoincrement=False)
     # Deterministic ID of exposure
 
     fid: Mapped[int] = mapped_column(ForeignKey("filters.fid"))
@@ -70,26 +66,26 @@ class ExposuresTable(SummerBase):  # pylint: disable=too-few-public-methods
 
     timeutc = Column(DateTime(timezone=True))
 
-    AExpTime = Column(Float, nullable=False)
-    expMJD = Column(Float, nullable=False)
+    aexptime = Column(Float, nullable=False)
+    expmjd = Column(Float, nullable=False)
     airmass = Column(Float)
     shutopen = Column(DateTime(timezone=True))
     shutclsd = Column(DateTime(timezone=True))
     tempture = Column(Float, default=-999)
     windspd = Column(Float, default=-999)
-    Dewpoint = Column(Float, default=-999)
-    Humidity = Column(Float, default=-999)
-    Pressure = Column(Float, default=-999)
-    Moonra = Column(Float, default=-999)
-    Moondec = Column(Float, default=-999)
-    Moonillf = Column(Float, default=-999)
-    Moonphas = Column(Float, default=-999)
-    Moonaz = Column(Float, default=-999)
-    Moonalt = Column(Float, default=-999)
-    Sunaz = Column(Float, default=-999)
-    Sunalt = Column(Float, default=-999)
-    Detsoft = Column(VARCHAR(50), default="unknown")
-    Detfirm = Column(VARCHAR(50), default="unknown")
+    dewpoint = Column(Float, default=-999)
+    humidity = Column(Float, default=-999)
+    pressure = Column(Float, default=-999)
+    moonra = Column(Float, default=-999)
+    moondec = Column(Float, default=-999)
+    moonillf = Column(Float, default=-999)
+    moonphas = Column(Float, default=-999)
+    moonaz = Column(Float, default=-999)
+    moonalt = Column(Float, default=-999)
+    sunaz = Column(Float, default=-999)
+    sunalt = Column(Float, default=-999)
+    detsoft = Column(VARCHAR(50), default="unknown")
+    detfirm = Column(VARCHAR(50), default="unknown")
     ra = Column(Float)
     dec = Column(Float)
     altitude = Column(Float)
@@ -99,16 +95,7 @@ class ExposuresTable(SummerBase):  # pylint: disable=too-few-public-methods
     dec_column_name = "dec"
 
     raw: Mapped["RawTable"] = relationship(back_populates="exposure_ids")
-
-
-# @event.listens_for(target=RawTable.__table__, identifier="after_create")
-# def raw_q3c(tbl, conn, *args, **kw):
-#     create_q3c_extension(
-#         conn=conn,
-#         __tablename__=RawTable.__tablename__,
-#         ra_column_name=RawTable.ra_column_name,
-#         dec_column_name=RawTable.dec_column_name,
-#     )
+    diff: Mapped["DiffTable"] = relationship(back_populates="exposure_ids")
 
 
 default_unknown_field = Field(default=-999)
@@ -129,32 +116,32 @@ class Exposure(BaseDB):
     puid: int = Field(ge=0)
 
     timeutc: datetime = Field()
-    AExpTime: float = Field(ge=0)
-    expMJD: float = Field(ge=59000)
+    aexptime: float = Field(ge=0)
+    expmjd: float = Field(ge=59000)
     airmass: float = Field(ge=1.0)
     shutopen: datetime = Field()
     shutclsd: datetime = Field()
     tempture: float = default_unknown_field
     windspd: float = default_unknown_field
-    Dewpoint: float = default_unknown_field
-    Humidity: float = default_unknown_field
-    Pressure: float = default_unknown_field
-    Moonra: float = Field(ge=0.0, le=360.0, default=None)
-    Moondec: float = Field(title="Dec (degrees)", ge=-90.0, le=90, default=None)
-    Moonillf: float = default_unknown_field
-    Moonphas: float = default_unknown_field
-    Moonaz: float = default_unknown_field
-    Moonalt: float = default_unknown_field
-    Sunaz: float = default_unknown_field
-    Sunalt: float = default_unknown_field
-    Detfirm: str = Field(default="unknown")
-    Detsoft: str = Field(default="unknown")
+    dewpoint: float = default_unknown_field
+    humidity: float = default_unknown_field
+    pressure: float = default_unknown_field
+    moonra: float = Field(ge=0.0, le=360.0, default=None)
+    moondec: float = Field(title="Dec (degrees)", ge=-90.0, le=90, default=None)
+    moonillf: float = default_unknown_field
+    moonphas: float = default_unknown_field
+    moonaz: float = default_unknown_field
+    moonalt: float = default_unknown_field
+    sunaz: float = default_unknown_field
+    sunalt: float = default_unknown_field
+    detfirm: str = Field(default="unknown")
+    detsoft: str = Field(default="unknown")
     ra: float = ra_field
     dec: float = dec_field
     altitude: float = alt_field
     azimuth: float = az_field
 
-    def insert_entry(self, returning_key_names=None) -> tuple:
+    def insert_entry(self, returning_key_names=None) -> pd.DataFrame:
         """
         Insert the pydantic-ified data into the corresponding sql database
 
@@ -165,14 +152,8 @@ def insert_entry(self, returning_key_names=None) -> tuple:
             night.insert_entry()
 
         logger.debug(f"puid: {self.puid}")
-        if not ProgramsTable().exists(values=self.puid, keys="puid"):
-            default_puid = ProgramsTable().select_query(
-                compare_values=default_program.progname,
-                compare_keys="progname",
-                select_keys="puid",
-            )[0][0]
-            logger.debug(f"Found progid {default_puid}")
-            self.puid = default_puid
+        if not Program._exists(values=self.puid, keys="puid"):
+            self.puid = 1
 
         return self._insert_entry()
 
@@ -182,4 +163,4 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.expid, keys="expid")
+        return self._exists(values=self.expid, keys="expid")
diff --git a/mirar/pipelines/summer/models/_fields.py b/mirar/pipelines/summer/models/_fields.py
index ff89fbb9e..34ea65db8 100644
--- a/mirar/pipelines/summer/models/_fields.py
+++ b/mirar/pipelines/summer/models/_fields.py
@@ -1,6 +1,7 @@
 """
 Models for the 'field' table
 """
+# pylint: disable=duplicate-code
 import time
 from typing import ClassVar
 
@@ -10,9 +11,10 @@
 from tqdm import tqdm
 from wintertoo.data import summer_fields
 
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.database.engine import get_engine
+from mirar.database.transactions import is_populated
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB, _exists, dec_field, ra_field
-from mirar.utils.sql import get_engine
 
 DEFAULT_FIELD = 999999999
 
@@ -59,7 +61,7 @@ def populate_fields():
     """
 
     engine = get_engine(db_name=FieldsTable.db_name)
-    if not _exists(Select(FieldsTable), engine=engine):
+    if not is_populated(FieldsTable):
         chunk = 10000
 
         summer_fields["fieldid"] = summer_fields["ID"]
diff --git a/mirar/pipelines/summer/models/_filters.py b/mirar/pipelines/summer/models/_filters.py
index 9e4ecaa55..fffabf4d5 100644
--- a/mirar/pipelines/summer/models/_filters.py
+++ b/mirar/pipelines/summer/models/_filters.py
@@ -1,14 +1,15 @@
 """
 Models for the 'filters' table
 """
+# pylint: disable=duplicate-code
 from typing import ClassVar
 
 from pydantic import Field, validator
 from sqlalchemy import VARCHAR, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB
 
 summer_filters_map = {"u": 1, "g": 2, "r": 3, "i": 4}
 fid_field: int = Field(ge=0)
@@ -42,7 +43,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.fid, keys="fid")
+        return self._exists(values=self.fid, keys="fid")
 
     @validator("fid")
     @classmethod
diff --git a/mirar/pipelines/summer/models/_img_type.py b/mirar/pipelines/summer/models/_img_type.py
index 7162335b3..c86930709 100644
--- a/mirar/pipelines/summer/models/_img_type.py
+++ b/mirar/pipelines/summer/models/_img_type.py
@@ -1,14 +1,15 @@
 """
 Models for the 'itid' table
 """
+# pylint: disable=duplicate-code
 from typing import ClassVar
 
 from pydantic import Field
 from sqlalchemy import VARCHAR, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB
 
 ALL_ITID = ["SCIENCE", "CAL", "FOCUS", "POINTING", "NULL"]
 DEFAULT_ITID = 5
@@ -43,7 +44,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.itid, keys="itid")
+        return self._exists(values=self.itid, keys="itid")
 
 
 def populate_itid():
diff --git a/mirar/pipelines/summer/models/_nights.py b/mirar/pipelines/summer/models/_nights.py
index 78cfd8cc9..db91c36c7 100644
--- a/mirar/pipelines/summer/models/_nights.py
+++ b/mirar/pipelines/summer/models/_nights.py
@@ -1,14 +1,15 @@
 """
 Models for the 'nights' table
 """
+# pylint: disable=duplicate-code
 from datetime import date
 from typing import ClassVar
 
 from sqlalchemy import DATE, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB, date_field
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB, date_field
 
 
 class NightsTable(SummerBase):  # pylint: disable=too-few-public-methods
@@ -38,4 +39,4 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.nightdate, keys="nightdate")
+        return self._exists(values=self.nightdate, keys="nightdate")
diff --git a/mirar/pipelines/summer/models/_proc.py b/mirar/pipelines/summer/models/_proc.py
index b0b84dc9d..260cd61ab 100644
--- a/mirar/pipelines/summer/models/_proc.py
+++ b/mirar/pipelines/summer/models/_proc.py
@@ -1,6 +1,7 @@
 """
 Models for the 'proc' table
 """
+# pylint: disable=duplicate-code
 import os
 from typing import ClassVar
 
@@ -8,9 +9,9 @@
 from sqlalchemy import REAL, VARCHAR, Column, ForeignKey, Integer, Sequence  # event,
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB, dec_field, ra_field
 from mirar.pipelines.summer.models._raw import Raw
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB, dec_field, ra_field
 
 
 class ProcTable(SummerBase):  # pylint: disable=too-few-public-methods
@@ -32,6 +33,9 @@ class ProcTable(SummerBase):  # pylint: disable=too-few-public-methods
     rawid: Mapped[int] = Column(Integer, ForeignKey("raw.rawid"), primary_key=True)
     raw_ids: Mapped["RawTable"] = relationship(back_populates="proc")
 
+    diffid: Mapped[int] = Column(Integer, ForeignKey("diff.diffid"))
+    diff_ids: Mapped["DiffTable"] = relationship(back_populates="proc")
+
     savepath = Column(VARCHAR(255), unique=True)
     wghtpath = Column(VARCHAR(255), unique=True)
 
@@ -104,5 +108,5 @@ def validate_expid(cls, field_value: int):
         Returns:
 
         """
-        assert Raw.sql_model().exists(keys="rawid", values=field_value)
+        assert Raw._exists(keys="rawid", values=field_value)
         return field_value
diff --git a/mirar/pipelines/summer/models/_programs.py b/mirar/pipelines/summer/models/_programs.py
index 168cb9647..e4363bc38 100644
--- a/mirar/pipelines/summer/models/_programs.py
+++ b/mirar/pipelines/summer/models/_programs.py
@@ -1,6 +1,7 @@
 """
 Models for the 'program' table
 """
+# pylint: disable=duplicate-code
 from datetime import date
 from typing import ClassVar
 
@@ -8,8 +9,8 @@
 from sqlalchemy import CHAR, DATE, REAL, VARCHAR, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB, date_field
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB, date_field
 from mirar.utils.security import generate_key
 
 LEN_PROG_KEY = 60
@@ -105,7 +106,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.progname, keys="progname")
+        return self._exists(values=self.progname, keys="progname")
 
 
 default_program = Program(
diff --git a/mirar/pipelines/summer/models/_raw.py b/mirar/pipelines/summer/models/_raw.py
index 009a73125..8e5c41f1f 100644
--- a/mirar/pipelines/summer/models/_raw.py
+++ b/mirar/pipelines/summer/models/_raw.py
@@ -1,6 +1,7 @@
 """
 Models for the 'raw' table
 """
+# pylint: disable=duplicate-code
 import os
 from typing import ClassVar
 
@@ -8,9 +9,9 @@
 from sqlalchemy import VARCHAR, Column, Double, ForeignKey, Integer, Sequence
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+from mirar.database.base_model import BaseDB
 from mirar.pipelines.summer.models._exposures import Exposure
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB
 
 
 class RawTable(SummerBase):  # pylint: disable=too-few-public-methods
@@ -77,5 +78,5 @@ def validate_expid(cls, field_value: int):
         Returns:
 
         """
-        assert Exposure.sql_model().exists(keys="uexpid", values=field_value)
+        assert Exposure._exists(keys="uexpid", values=field_value)
         return field_value
diff --git a/mirar/pipelines/summer/models/_subdets.py b/mirar/pipelines/summer/models/_subdets.py
index 5e567f19e..824a9d895 100644
--- a/mirar/pipelines/summer/models/_subdets.py
+++ b/mirar/pipelines/summer/models/_subdets.py
@@ -1,15 +1,16 @@
 """
 Models for the 'subdets' table
 """
+# pylint: disable=duplicate-code
 from typing import ClassVar
 
 from pydantic import Field
-from sqlalchemy import Column, Integer, Select
+from sqlalchemy import Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB
+from mirar.database.transactions import is_populated
 from mirar.pipelines.summer.models.base_model import SummerBase
-from mirar.processors.sqldatabase.base_model import BaseDB, _exists
-from mirar.utils.sql import get_engine
 
 DEFAULT_FIELD = 999999999
 
@@ -50,10 +51,12 @@ def populate_subdets(ndetectors: int = 1, nxtot: int = 1, nytot: int = 1):
     """
     Creates entries in the database based on number of detectors and splits
 
+    :param ndetectors: Number of detectors
+    :param nxtot: Number of splits in x direction
+    :param nytot: Number of splits in y direction
+    :return: None
     """
-
-    engine = get_engine(db_name=SubdetsTable.db_name)
-    if not _exists(Select(SubdetsTable), engine=engine):
+    if not is_populated(SubdetsTable):
         for ndetector in range(ndetectors):
             for nx in range(nxtot):
                 for ny in range(nytot):
@@ -64,9 +67,4 @@ def populate_subdets(ndetectors: int = 1, nxtot: int = 1, nytot: int = 1):
                         nxtot=nxtot,
                         nytot=nytot,
                     )
-
-                    if not new.sql_model().exists(
-                        values=[nx, nxtot, ny, nytot],
-                        keys=["nx", "nxtot", "ny", "nytot"],
-                    ):
-                        new.insert_entry()
+                    new.insert_entry()
diff --git a/mirar/pipelines/summer/models/base_model.py b/mirar/pipelines/summer/models/base_model.py
index d8a4631aa..b87eba803 100644
--- a/mirar/pipelines/summer/models/base_model.py
+++ b/mirar/pipelines/summer/models/base_model.py
@@ -3,7 +3,7 @@
 """
 from sqlalchemy.orm import DeclarativeBase
 
-from mirar.processors.sqldatabase.base_model import BaseTable
+from mirar.database.base_table import BaseTable
 
 DB_NAME = "summer"
 
diff --git a/mirar/pipelines/summer/summer_files/schema/q3c/q3c_indexes.sql b/mirar/pipelines/summer/summer_files/schema/q3c/q3c_indexes.sql
deleted file mode 100644
index ce79a0f1d..000000000
--- a/mirar/pipelines/summer/summer_files/schema/q3c/q3c_indexes.sql
+++ /dev/null
@@ -1,9 +0,0 @@
-CREATE EXTENSION q3c;
-
-CREATE INDEX IF NOT EXISTS raw_q3c_ang2ipix_idx ON raw (q3c_ang2ipix(ra, dec));
-CLUSTER raw_q3c_ang2ipix_idx ON raw;
-ANALYZE raw;
-
-CREATE INDEX IF NOT EXISTS proc_q3c_ang2ipix_idx ON proc (q3c_ang2ipix(crval1, crval2));
-CLUSTER proc_q3c_ang2ipix_idx ON proc;
-ANALYZE proc;
diff --git a/mirar/pipelines/winter/blocks.py b/mirar/pipelines/winter/blocks.py
index c9ca7ace1..061cbfe71 100644
--- a/mirar/pipelines/winter/blocks.py
+++ b/mirar/pipelines/winter/blocks.py
@@ -1,10 +1,12 @@
 """
 Module for WINTER data reduction
 """
+# pylint: disable=duplicate-code
 from mirar.catalog.kowalski import PS1, TMASS
 from mirar.downloader.get_test_data import get_test_data_dir
 from mirar.paths import (
     BASE_NAME_KEY,
+    CAND_NAME_KEY,
     DITHER_N_KEY,
     EXPTIME_KEY,
     MAX_DITHER_KEY,
@@ -14,6 +16,7 @@
     base_output_dir,
 )
 from mirar.pipelines.winter.config import (
+    prv_candidate_cols,
     psfex_path,
     scamp_config_path,
     sextractor_anet_config,
@@ -24,8 +27,8 @@
     sextractor_reference_config,
     swarp_config_path,
     winter_avro_schema_path,
-    winter_candidate_config,
 )
+from mirar.pipelines.winter.constants import NXSPLIT, NYSPLIT
 from mirar.pipelines.winter.generator import (
     winter_astrometric_ref_catalog_generator,
     winter_astrometry_sextractor_catalog_purifier,
@@ -46,10 +49,12 @@
     load_winter_mef_image,
 )
 from mirar.pipelines.winter.models import (
+    CANDIDATE_PREFIX,
     DEFAULT_FIELD,
-    NXSPLIT,
-    NYSPLIT,
+    NAME_START,
     AstrometryStat,
+    Candidate,
+    Diff,
     Exposure,
     Raw,
     Stack,
@@ -65,10 +70,13 @@
 from mirar.processors.avro import IPACAvroExporter
 from mirar.processors.csvlog import CSVLog
 from mirar.processors.dark import DarkCalibrator
-from mirar.processors.database.database_exporter import DatabaseDataframeExporter
-from mirar.processors.database.database_importer import DatabaseHistoryImporter
-from mirar.processors.database.database_modifier import ModifyImageDatabaseSeqList
-from mirar.processors.database.utils import get_column_names_from_schema
+from mirar.processors.database.database_inserter import (
+    DatabaseImageBatchInserter,
+    DatabaseImageInserter,
+    DatabaseSourceInserter,
+)
+from mirar.processors.database.database_selector import DatabaseHistorySelector
+from mirar.processors.database.database_updater import ImageDatabaseMultiEntryUpdater
 from mirar.processors.mask import (  # MaskAboveThreshold,
     MaskDatasecPixels,
     MaskPixelsFromFunction,
@@ -85,10 +93,6 @@
     SourceWriter,
 )
 from mirar.processors.split import SUB_ID_KEY, SplitImage
-from mirar.processors.sqldatabase.database_exporter import (
-    DatabaseImageBatchExporter,
-    DatabaseImageExporter,
-)
 from mirar.processors.utils import (
     CustomImageModifier,
     HeaderAnnotator,
@@ -151,7 +155,7 @@
 
 extract_all = [
     ImageBatcher("UTCTIME"),
-    DatabaseImageBatchExporter(db_table=Exposure, duplicate_protocol="ignore"),
+    DatabaseImageBatchInserter(db_table=Exposure, duplicate_protocol="ignore"),
     ImageSelector((OBSCLASS_KEY, ["dark", "science"])),
 ]
 
@@ -213,7 +217,7 @@
 
 save_raw = [
     ImageSaver(output_dir_name="raw_unpacked", write_mask=False),
-    DatabaseImageExporter(db_table=Raw, duplicate_protocol="replace", q3c_bool=False),
+    DatabaseImageInserter(db_table=Raw, duplicate_protocol="replace"),
 ]
 
 # +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
@@ -300,7 +304,7 @@
         cache=True,
         crossmatch_radius_arcsec=5.0,
     ),
-    DatabaseImageExporter(db_table=AstrometryStat, duplicate_protocol="ignore"),
+    DatabaseImageInserter(db_table=AstrometryStat, duplicate_protocol="ignore"),
 ]
 
 stack_dithers = [
@@ -335,12 +339,10 @@
         cache=True,
     ),
     ImageSaver(output_dir_name="final"),
-    DatabaseImageExporter(db_table=Stack, duplicate_protocol="replace", q3c_bool=False),
-    ModifyImageDatabaseSeqList(
-        db_name="winter",
-        schema_path="fake_placeholder_path.sql",
+    DatabaseImageInserter(db_table=Stack, duplicate_protocol="replace"),
+    ImageDatabaseMultiEntryUpdater(
         sequence_key="rawid",
-        db_table=Raw.sql_model.__tablename__,
+        db_table=Raw,
         db_alter_columns="ustackid",
     ),
 ]
@@ -373,7 +375,7 @@
     ZOGY(
         output_sub_dir="subtract", sci_zp_header_key="ZP_AUTO", ref_zp_header_key=ZP_KEY
     ),
-    ImageSaver(output_dir_name="diffs"),
+    DatabaseImageInserter(db_table=Diff, duplicate_protocol="replace"),
 ]
 
 detect_candidates = [
@@ -393,8 +395,8 @@
     ),
     SourceWriter(output_dir_name="candidates"),
 ]
-
-candidate_colnames = get_column_names_from_schema(winter_candidate_config)
+#
+# candidate_colnames = get_column_names_from_schema(winter_candidate_config)
 
 load_candidates = [
     SourceLoader(input_dir_name="candidates"),
@@ -405,29 +407,22 @@
     XMatch(catalog=TMASS(num_sources=3, search_radius_arcmin=0.5)),
     XMatch(catalog=PS1(num_sources=3, search_radius_arcmin=0.5)),
     SourceWriter(output_dir_name="kowalski"),
-    DatabaseHistoryImporter(
+    DatabaseHistorySelector(
         crossmatch_radius_arcsec=2.0,
         time_field_name="jd",
         history_duration_days=500.0,
-        db_name="winter",
-        db_table="candidates",
-        db_output_columns=candidate_colnames,
-        schema_path=winter_candidate_config,
-        q3c_bool=False,
+        db_table=Candidate,
+        db_output_columns=prv_candidate_cols + [CAND_NAME_KEY],
     ),
     CandidateNamer(
-        db_name="winter",
-        db_table="candidates",
-        base_name="WNTR",
-        name_start="aaaaa",
+        db_table=Candidate,
+        base_name=CANDIDATE_PREFIX,
+        name_start=NAME_START,
         xmatch_radius_arcsec=2,
-        schema_path=winter_candidate_config,
     ),
-    DatabaseDataframeExporter(
-        db_name="winter",
-        db_table="candidates",
-        schema_path=winter_candidate_config,
-        duplicate_protocol="replace",
+    DatabaseSourceInserter(
+        db_table=Candidate,
+        duplicate_protocol="fail",
     ),
     SourceWriter(output_dir_name="preavro"),
     IPACAvroExporter(
diff --git a/mirar/pipelines/winter/config/__init__.py b/mirar/pipelines/winter/config/__init__.py
index 2dcda82e9..daffcab20 100644
--- a/mirar/pipelines/winter/config/__init__.py
+++ b/mirar/pipelines/winter/config/__init__.py
@@ -3,6 +3,8 @@
 """
 from pathlib import Path
 
+from fastavro.schema import load_schema
+
 from mirar.processors.utils.cal_hunter import CalRequirement
 
 PIPELINE_NAME = "winter"
@@ -54,8 +56,6 @@
 scamp_config_path = winter_file_dir.joinpath("astrom.scamp")
 winter_mask_path = winter_file_dir.joinpath("winter_mask.fits")
 
-winter_candidate_config = winter_file_dir.joinpath("candidates.sql")
-
 psfex_path = winter_file_dir.joinpath("photom.psfex")
 
 winter_cal_requirements = [
@@ -65,3 +65,6 @@
 ]
 
 winter_avro_schema_path = winter_file_dir.joinpath("avro_schema/winter.alert.avsc")
+winter_avro_schema = load_schema(winter_avro_schema_path)
+winter_prv_schema = winter_avro_schema["__named_schemas"]["winter.alert.prv_candidate"]
+prv_candidate_cols = [x["name"] for x in winter_prv_schema["fields"]]
diff --git a/mirar/pipelines/winter/config/files/avro_schema/winter.alert.avsc b/mirar/pipelines/winter/config/files/avro_schema/winter.alert.avsc
index 5dc7de436..abefbd8da 100644
--- a/mirar/pipelines/winter/config/files/avro_schema/winter.alert.avsc
+++ b/mirar/pipelines/winter/config/files/avro_schema/winter.alert.avsc
@@ -6,14 +6,14 @@
     "fields": [
         {"name": "schemavsn", "type": "string", "doc": "schema version used", "default": "0.1"},
         {"name": "publisher", "type": "string", "doc": "origin of alert packet", "default": "WINTER-mirar"},
-        {"name": "objectId", "type": "string"},
+        {"name": "objectid", "type": "string"},
         {"name": "candid", "type": "long"},
         {"name": "candidate", "type": "winter.alert.candidate", "doc": "candidate information"},
         {"name": "prv_candidates", "type": [{
             "type": "array",
             "items": "winter.alert.prv_candidate"}, "null"]},
-        {"name": "cutoutScience", "type": "bytes", "doc": "science cutout image [gzipped fits bytestream]"},
-        {"name": "cutoutTemplate", "type": "bytes", "doc": "cutout of template image [gzipped fits bytestream]"},
-        {"name": "cutoutDifference", "type": "bytes", "doc": "difference image cutout [gzipped fits bytestream]"}
+        {"name": "cutout_science", "type": "bytes", "doc": "science cutout image [gzipped fits bytestream]"},
+        {"name": "cutout_template", "type": "bytes", "doc": "cutout of template image [gzipped fits bytestream]"},
+        {"name": "cutout_difference", "type": "bytes", "doc": "difference image cutout [gzipped fits bytestream]"}
         ]
 }
diff --git a/mirar/pipelines/winter/config/files/avro_schema/winter.alert.candidate.avsc b/mirar/pipelines/winter/config/files/avro_schema/winter.alert.candidate.avsc
index 0204ec80f..c45b6aeb4 100644
--- a/mirar/pipelines/winter/config/files/avro_schema/winter.alert.candidate.avsc
+++ b/mirar/pipelines/winter/config/files/avro_schema/winter.alert.candidate.avsc
@@ -4,35 +4,42 @@
 	"name": "candidate",
 	"doc": "avro alert candidate schema for WINTER",
 	"fields": [
+		{"name": "candid", "type": "long", "doc": "Candidate ID from candidates DB"},
+		{"name": "deprecated", "type": "boolean", "doc": "Whether this candidate is superceded by a reprocessing"},
 		{"name": "jd", "type": "double", "doc": "Observation Julian date at start of exposure [days]"},
-		{"name": "diffmaglim", "type": ["float", "null"], "doc": "Expected 5-sigma mag limit in difference image based on global noise estimate [mag]"},
 		{"name": "fid", "type": "int", "doc": "Filter ID (0=Y; 1=J; 2=H; 3=K)"},
-		{"name": "programpi", "type": ["string", "null"], "doc": "Principal investigator attached to program ID"},
-		{"name": "programid", "type": "int", "doc": "Program ID: encodes either public, collab, or caltech mode"},
-		{"name": "candid", "type": "long", "doc": "Candidate ID from operations DB"},
-		{"name": "isdiffpos", "type": ["float"], "doc": "t or 1 => candidate is from positive (sci minus ref) subtraction; f or 0 => candidate is from negative (ref minus sci) subtraction"},
-		{"name": "field", "type": ["int", "null"], "doc": "ZTF field ID"},
-		{"name": "ndethist", "type": ["int"], "default": 1, "doc": "Number of detections"},
-		{"name": "xpos", "type": ["float", "null"], "doc": "x-image position of candidate [pixels]"},
-		{"name": "ypos", "type": ["float", "null"], "doc": "y-image position of candidate [pixels]"},
+		{"name": "exptime", "type": "float", "doc": "Integration time of camera exposure [sec]"},
+		{"name": "ndethist", "type": "int", "default": 1, "doc": "Number of detections"},
+		{"name": "jdstarthist", "type": "double", "doc": "Earliest Julian date of epoch corresponding to ndethist [days]"},
+        {"name": "jdendhist", "type": "double", "doc": "Latest Julian date of epoch corresponding to ndethist [days]"},
+		{"name": "progname", "type": "string", "doc": "Name of program e.g 2023A000"},
+		{"name": "programid", "type": "int", "doc": "Program ID: encodes either public, collab, MIT or caltech mode"},
+		{"name": "isdiffpos", "type": "boolean", "doc": "boolean true => candidate is from positive (sci minus ref) subtraction; false => candidate is from negative (ref minus sci) subtraction"},
+		{"name": "field", "type": "int", "doc": "WINTER field ID"},
 		{"name": "ra", "type": "double", "doc": "Right Ascension of candidate; J2000 [deg]"},
-		{"name": "dec", "type": "double",  "doc": "Declination of candidate; J2000 [deg]"},		
+		{"name": "dec", "type": "double",  "doc": "Declination of candidate; J2000 [deg]"},
+		{"name": "magzpsci", "type": ["float", "null"], "doc": "Magnitude zero point for photometry estimates [mag]"},
+		{"name": "magzpsciunc", "type": ["float", "null"], "doc": "Magnitude zero point uncertainty (in magzpsci) [mag]"},
+		{"name": "magzpscirms", "type": ["float", "null"], "doc": "RMS (deviation from average) in all differences between instrumental photometry and matched photometric calibrators from science image processing [mag]"},
+		{"name": "diffmaglim", "type": ["float", "null"], "doc": "Expected 5-sigma mag limit in difference image based on global noise estimate [mag]"},
 		{"name": "magpsf", "type": "float",   "doc": "Magnitude from PSF-fit photometry [mag]"},
 		{"name": "sigmapsf", "type": "float",  "doc": "1-sigma uncertainty in magpsf [mag]"},
 		{"name": "chipsf", "type": ["float", "null"], "doc": "Reduced chi-square for PSF-fit"},
 		{"name": "magap", "type": ["float", "null"], "doc": "Aperture mag using 14 pixel diameter aperture [mag]"},
 		{"name": "sigmagap", "type": ["float", "null"], "doc": "1-sigma uncertainty in magap [mag]"},
+		{"name": "magapbig", "type": ["float", "null"], "doc": "Aperture mag using 18 pixel diameter aperture [mag]"},
+		{"name": "sigmagapbig", "type": ["float", "null"], "doc": "1-sigma uncertainty in magapbig [mag]"},
+		{"name": "magdiff", "type": ["float", "null"], "doc": "Difference: magap - magpsf [mag]"},
+		{"name": "magfromlim", "type": ["float", "null"], "doc": "Difference: diffmaglim - magap [mag]"},
 		{"name": "distnr", "type": ["float", "null"], "doc": "distance to nearest source in reference image PSF-catalog [pixels]"},
 		{"name": "magnr", "type": ["float", "null"], "doc": "magnitude of nearest source in reference image PSF-catalog [mag]"},
 		{"name": "sigmagnr", "type": ["float", "null"], "doc": "1-sigma uncertainty in magnr [mag]"},
-		{"name": "chinr", "type": ["float", "null"], "doc": "DAOPhot chi parameter of nearest source in reference image PSF-catalog"},
-		{"name": "sharpnr", "type": ["float", "null"], "doc": "DAOPhot sharp parameter of nearest source in reference image PSF-catalog"},
+		{"name": "xpos", "type": ["float", "null"], "doc": "x-image position of candidate [pixels]"},
+		{"name": "ypos", "type": ["float", "null"], "doc": "y-image position of candidate [pixels]"},
 		{"name": "sky", "type": ["float", "null"], "doc": "Local sky background estimate [DN]"},
-		{"name": "magdiff", "type": ["float", "null"], "doc": "Difference: magap - magpsf [mag]"},
 		{"name": "fwhm", "type": ["float", "null"], "doc": "Full Width Half Max assuming a Gaussian core, from SExtractor [pixels]"},
 		{"name": "classtar", "type": ["float", "null"], "doc": "Star/Galaxy classification score from SExtractor"},
 		{"name": "mindtoedge", "type": ["float", "null"], "doc": "Distance to nearest edge in image [pixels]"},
-		{"name": "magfromlim", "type": ["float", "null"], "doc": "Difference: diffmaglim - magap [mag]"},
 		{"name": "seeratio", "type": ["float", "null"], "doc": "Ratio: difffwhm / fwhm"},
 		{"name": "aimage", "type": ["float", "null"], "doc": "Windowed profile RMS afloat major axis from SExtractor [pixels]"},
 		{"name": "bimage", "type": ["float", "null"], "doc": "Windowed profile RMS afloat minor axis from SExtractor [pixels]"},
@@ -41,29 +48,31 @@
 		{"name": "elong", "type": ["float", "null"], "doc": "Ratio: aimage / bimage"},
 		{"name": "nneg", "type": ["int", "null"], "doc": "number of negative pixels in a 5 x 5 pixel stamp"},
 		{"name": "nbad", "type": ["int", "null"], "doc": "number of prior-tagged bad pixels in a 5 x 5 pixel stamp"},
+		{"name": "sumrat", "type": ["float", "null"], "doc": "Ratio: sum(pixels) / sum(|pixels|) in a 5 x 5 pixel stamp where stamp is first median-filtered to mitigate outliers"},
+		{"name": "dsnrms", "type": ["float", "null"], "doc": "Ratio: D/stddev(D) on event position where D = difference image"},
+		{"name": "ssnrms", "type": ["float", "null"], "doc": "Ratio: S/stddev(S) on event position where S = image of convolution: D (x) PSF(D)"},
+		{"name": "dsdiff", "type": ["float", "null"], "doc": "Difference of statistics: dsnrms - ssnrms"},
+		{"name": "scorr", "type": ["double", "null"], "doc": "Peak-pixel signal-to-noise ratio in point source matched-filtered detection image"},
+		{"name": "clrcoeff", "type": ["float", "null"], "doc": "Color coefficient from linear fit from photometric calibration of science image"},
+		{"name": "clrcounc", "type": ["float", "null"], "doc": "Color coefficient uncertainty from linear fit (corresponding to clrcoeff)"},
+		{"name": "zpclrcov", "type": ["float", "null"], "doc": "Covariance in magzpsci and clrcoeff from science image processing [mag^2]"},
+		{"name": "zpmed", "type": ["float", "null"], "doc": "Magnitude zero point from median of all differences between instrumental photometry and matched photometric calibrators from science image processing [mag]"},
+		{"name": "clrmed", "type": ["float", "null"], "doc": "Median color of all PS1 photometric calibrators used from science image processing [mag]: for filter (fid) = 1, 2, 3, PS1 color used = g-r, g-r, r-i respectively"},
+		{"name": "clrrms", "type": ["float", "null"], "doc": "RMS color (deviation from average) of all PS1 photometric calibrators used from science image processing [mag]"},
 		{"name": "rb", "type": ["float", "null"], "doc": "RealBogus quality score from Random Forest classifier; range is 0 to 1 where closer to 1 is more reliable"},
 		{"name": "rbversion", "type": ["string", "null"], "doc": "version of Random Forest classifier model used to assign RealBogus (rb) quality score"},
-        {"name": "ssdistnr", "type": ["float", "null"], "doc": "distance to nearest known solar system object if exists within 30 arcsec [arcsec]"},
+		{"name": "ssdistnr", "type": ["float", "null"], "doc": "distance to nearest known solar system object if exists within 30 arcsec [arcsec]"},
 		{"name": "ssmagnr", "type": ["float", "null"], "doc": "magnitude of nearest known solar system object if exists within 30 arcsec (usually V-band from MPC archive) [mag]"},
 		{"name": "ssnamenr", "type": ["string", "null"], "doc": "name of nearest known solar system object if exists within 30 arcsec (from MPC archive)"},
-		{"name": "sumrat", "type": ["float", "null"], "doc": "Ratio: sum(pixels) / sum(|pixels|) in a 5 x 5 pixel stamp where stamp is first median-filtered to mitigate outliers"},
-		{"name": "magapbig", "type": ["float", "null"], "doc": "Aperture mag using 18 pixel diameter aperture [mag]"},
-		{"name": "sigmagapbig", "type": ["float", "null"], "doc": "1-sigma uncertainty in magapbig [mag]"},
+        {"name": "tooflag", "type": "boolean", "default": false, "doc": "true => candidate is from a Target-of-Opportunity (ToO) exposure; false => candidate is from a non-ToO exposure"},
+		{"name": "nmtchps", "type": "int", "doc": "Number of source matches from PS1 catalog falling within 30 arcsec"},
 		{"name": "psra1", "type": "double",  "doc": "Right Ascension of nearest source in reference image PSF-catalog; J2000 [deg]"},
 		{"name": "psdec1", "type": "double",  "doc": "Declination of nearest source in reference image PSF-catalog; J2000 [deg]"},
-        {"name": "jdstarthist", "type": ["double", "null"], "doc": "Earliest Julian date of epoch corresponding to ndethist [days]"},
-        {"name": "jdendhist", "type": ["double", "null"], "doc": "Latest Julian date of epoch corresponding to ndethist [days]"},
-        {"name": "scorr", "type": ["double", "null"], "doc": "Peak-pixel signal-to-noise ratio in point source matched-filtered detection image"},
-        {"name": "tooflag", "type": ["int", "null"], "default": 0, "doc": "1 => candidate is from a Target-of-Opportunity (ToO) exposure; 0 => candidate is from a non-ToO exposure"},
-        {"name": "psobjectid1", "type": ["float", "null"], "doc": "Object ID of closest source from PS1 catalog; if exists within 30 arcsec"},
-        {"name": "sgmag1", "type": ["float", "null"], "doc": "g-band PSF-fit magnitude of closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
+		{"name": "psobjectid1", "type": ["float", "null"], "doc": "Object ID of closest source from PS1 catalog; if exists within 30 arcsec"},        {"name": "sgmag1", "type": ["float", "null"], "doc": "g-band PSF-fit magnitude of closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
+		{"name": "sgmag1", "type": ["float", "null"], "doc": "g-band PSF-fit magnitude of second closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "srmag1", "type": ["float", "null"], "doc": "r-band PSF-fit magnitude of closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "simag1", "type": ["float", "null"], "doc": "i-band PSF-fit magnitude of closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "szmag1", "type": ["float", "null"], "doc": "z-band PSF-fit magnitude of closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
-		{"name": "tmjmag1", "type": ["float", "null"], "doc": "J-band PSF magnitude of closest source from 2MASS catalog; if exists within 30 arcsec [mag]"},
-		{"name": "tmhmag1", "type": ["float", "null"], "doc": "H-band PSF magnitude of closest source from 2MASS catalog; if exists within 30 arcsec [mag]"},
-		{"name": "tmkmag1", "type": ["float", "null"], "doc": "K-band PSF magnitude of closest source from 2MASS catalog; if exists within 30 arcsec [mag]"},
-        {"name": "tmobjectid1", "type": ["string", "null"], "doc": "Object ID of closest source from 2MASS catalog; if exists within 30 arcsec"},
 		{"name": "sgscore1", "type": ["float", "null"], "doc": "Star/Galaxy score of closest source from PS1 catalog; if exists within 30 arcsec: 0 <= sgscore <= 1 where closer to 1 implies higher likelihood of being a star"},
 		{"name": "distpsnr1", "type": ["float", "null"], "doc": "Distance to closest source from PS1 catalog; if exists within 30 arcsec [arcsec]"},
 		{"name": "psobjectid2", "type": ["float", "null"], "doc": "Object ID of second closest source from PS1 catalog; if exists within 30 arcsec"},
@@ -73,39 +82,29 @@
 		{"name": "szmag2", "type": ["float", "null"], "doc": "z-band PSF-fit magnitude of second closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "sgscore2", "type": ["float", "null"], "doc": "Star/Galaxy score of second closest source from PS1 catalog; if exists within 30 arcsec: 0 <= sgscore <= 1 where closer to 1 implies higher likelihood of being a star"},
 		{"name": "distpsnr2", "type": ["float", "null"], "doc": "Distance to second closest source from PS1 catalog; if exists within 30 arcsec [arcsec]"},
-        {"name": "tmjmag2", "type": ["float", "null"], "doc": "J-band PSF magnitude of second source from 2MASS catalog; if exists within 30 arcsec [mag]"},
-        {"name": "tmhmag2", "type": ["float", "null"], "doc": "H-band PSF magnitude of second source from 2MASS catalog; if exists within 30 arcsec [mag]"},
-        {"name": "tmkmag2", "type": ["float", "null"], "doc": "K-band PSF magnitude of second source from 2MASS catalog; if exists within 30 arcsec [mag]"},
-        {"name": "tmobjectid2", "type": ["string", "null"], "doc": "Object ID of second source from 2MASS catalog; if exists within 30 arcsec"},
-        {"name": "psobjectid3", "type": ["float", "null"], "doc": "Object ID of third closest source from PS1 catalog; if exists within 30 arcsec"},            
+        {"name": "psobjectid3", "type": ["float", "null"], "doc": "Object ID of third closest source from PS1 catalog; if exists within 30 arcsec"},
 		{"name": "sgmag3", "type": ["float", "null"], "doc": "g-band PSF-fit magnitude of third closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "srmag3", "type": ["float", "null"], "doc": "r-band PSF-fit magnitude of third closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "simag3", "type": ["float", "null"], "doc": "i-band PSF-fit magnitude of third closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "szmag3", "type": ["float", "null"], "doc": "z-band PSF-fit magnitude of third closest source from PS1 catalog; if exists within 30 arcsec [mag]"},
 		{"name": "sgscore3", "type": ["float", "null"], "doc": "Star/Galaxy score of third closest source from PS1 catalog; if exists within 30 arcsec: 0 <= sgscore <= 1 where closer to 1 implies higher likelihood of being a star"},
 		{"name": "distpsnr3", "type": ["float", "null"], "doc": "Distance to third closest source from PS1 catalog; if exists within 30 arcsec [arcsec]"},
-        {"name": "tmjmag3", "type": ["float", "null"], "doc": "J-band PSF magnitude of third source from 2MASS catalog; if exists within 30 arcsec [mag]"},                
+		{"name": "nmtchtm", "type": "int", "doc": "Number of source matches from 2MASS catalog falling within 30 arcsec"},
+		{"name": "tmjmag1", "type": ["float", "null"], "doc": "J-band PSF magnitude of closest source from 2MASS catalog; if exists within 30 arcsec [mag]"},
+		{"name": "tmhmag1", "type": ["float", "null"], "doc": "H-band PSF magnitude of closest source from 2MASS catalog; if exists within 30 arcsec [mag]"},
+		{"name": "tmkmag1", "type": ["float", "null"], "doc": "K-band PSF magnitude of closest source from 2MASS catalog; if exists within 30 arcsec [mag]"},
+        {"name": "tmobjectid1", "type": ["string", "null"], "doc": "Object ID of closest source from 2MASS catalog; if exists within 30 arcsec"},
+		{"name": "tmjmag2", "type": ["float", "null"], "doc": "J-band PSF magnitude of second source from 2MASS catalog; if exists within 30 arcsec [mag]"},
+        {"name": "tmhmag2", "type": ["float", "null"], "doc": "H-band PSF magnitude of second source from 2MASS catalog; if exists within 30 arcsec [mag]"},
+        {"name": "tmkmag2", "type": ["float", "null"], "doc": "K-band PSF magnitude of second source from 2MASS catalog; if exists within 30 arcsec [mag]"},
+        {"name": "tmobjectid2", "type": ["string", "null"], "doc": "Object ID of second source from 2MASS catalog; if exists within 30 arcsec"},
+        {"name": "tmjmag3", "type": ["float", "null"], "doc": "J-band PSF magnitude of third source from 2MASS catalog; if exists within 30 arcsec [mag]"},
         {"name": "tmhmag3", "type": ["float", "null"], "doc": "H-band PSF magnitude of third source from 2MASS catalog; if exists within 30 arcsec [mag]"},
         {"name": "tmkmag3", "type": ["float", "null"], "doc": "K-band PSF magnitude of third source from 2MASS catalog; if exists within 30 arcsec [mag]"},
         {"name": "tmobjectid3", "type": ["string", "null"], "doc": "Object ID of third source from 2MASS catalog; if exists within 30 arcsec"},
-        {"name": "nmtchps", "type": "int", "doc": "Number of source matches from PS1 catalog falling within 30 arcsec"},
-        {"name": "nmtchtm", "type": "int", "doc": "Number of source matches from 2MASS catalog falling within 30 arcsec"},
-		{"name": "dsnrms", "type": ["float", "null"], "doc": "Ratio: D/stddev(D) on event position where D = difference image"},
-		{"name": "ssnrms", "type": ["float", "null"], "doc": "Ratio: S/stddev(S) on event position where S = image of convolution: D (x) PSF(D)"},
-		{"name": "dsdiff", "type": ["float", "null"], "doc": "Difference of statistics: dsnrms - ssnrms"},
-		{"name": "magzpsci", "type": ["float", "null"], "doc": "Magnitude zero point for photometry estimates [mag]"},
-		{"name": "magzpsciunc", "type": ["float", "null"], "doc": "Magnitude zero point uncertainty (in magzpsci) [mag]"}, 
-		{"name": "magzpscirms", "type": ["float", "null"], "doc": "RMS (deviation from average) in all differences between instrumental photometry and matched photometric calibrators from science image processing [mag]"}, 
-		{"name": "clrcoeff", "type": ["float", "null"], "doc": "Color coefficient from linear fit from photometric calibration of science image"}, 
-		{"name": "clrcounc", "type": ["float", "null"], "doc": "Color coefficient uncertainty from linear fit (corresponding to clrcoeff)"},
-		{"name": "zpclrcov", "type": ["float", "null"], "doc": "Covariance in magzpsci and clrcoeff from science image processing [mag^2]"},
-		{"name": "zpmed", "type": ["float", "null"], "doc": "Magnitude zero point from median of all differences between instrumental photometry and matched photometric calibrators from science image processing [mag]"},
-		{"name": "clrmed", "type": ["float", "null"], "doc": "Median color of all PS1 photometric calibrators used from science image processing [mag]: for filter (fid) = 1, 2, 3, PS1 color used = g-r, g-r, r-i respectively"},
-		{"name": "clrrms", "type": ["float", "null"], "doc": "RMS color (deviation from average) of all PS1 photometric calibrators used from science image processing [mag]"},
 		{"name": "neargaia", "type": ["float", "null"], "doc": "Distance to closest source from Gaia DR1 catalog irrespective of magnitude; if exists within 90 arcsec [arcsec]"},
 		{"name": "neargaiabright", "type": ["float", "null"], "doc": "Distance to closest source from Gaia DR1 catalog brighter than magnitude 14; if exists within 90 arcsec [arcsec]"},
 		{"name": "maggaia", "type": ["float", "null"], "doc": "Gaia (G-band) magnitude of closest source from Gaia DR1 catalog irrespective of magnitude; if exists within 90 arcsec [mag]"},
-		{"name": "maggaiabright", "type": ["float", "null"], "doc": "Gaia (G-band) magnitude of closest source from Gaia DR1 catalog brighter than magnitude 14; if exists within 90 arcsec [mag]"},
-		{"name": "exptime", "type": ["float", "null"], "doc": "Integration time of camera exposure [sec]"}
+		{"name": "maggaiabright", "type": ["float", "null"], "doc": "Gaia (G-band) magnitude of closest source from Gaia DR1 catalog brighter than magnitude 14; if exists within 90 arcsec [mag]"}
 			]
 }
diff --git a/mirar/pipelines/winter/config/files/avro_schema/winter.alert.prv_candidate.avsc b/mirar/pipelines/winter/config/files/avro_schema/winter.alert.prv_candidate.avsc
index aeba7ebb5..2bca0877e 100644
--- a/mirar/pipelines/winter/config/files/avro_schema/winter.alert.prv_candidate.avsc
+++ b/mirar/pipelines/winter/config/files/avro_schema/winter.alert.prv_candidate.avsc
@@ -5,17 +5,15 @@
     "type": "record",
     "fields": [
 		{"name": "candid", "type": ["long", "null"], "doc": "Candidate ID from operations DB"},
+		{"name": "progname", "type": "string", "doc": "Name of program e.g 2023A000"},
+		{"name": "jd", "type": "double", "doc": "Observation Julian date at start of exposure [days]"},
+		{"name": "fid", "type": "int", "doc": "Filter ID (0=Y; 1=J; 2=H; 3=K)"},
 		{"name": "isdiffpos", "type": ["string", "null"], "doc": "t or 1 => candidate is from positive (sci minus ref) subtraction; f or 0 => candidate is from negative (ref minus sci) subtraction"},
-		{"name": "nid", "type": ["int", "null"], "doc": "Night ID"},
-		{"name": "quadpos", "type": ["int", "null"], "doc": "Quadrant ID [0..3]"},
-		{"name": "subquadpos", "type": ["int", "null"], "doc": "Sub-Quadrant ID [0..3]"},
-		{"name": "field", "type": ["int", "null"], "doc": "field ID"},
-		{"name": "xpos", "type": ["float", "null"], "doc": "x-image position of candidate [pixels]"},
-		{"name": "ypos", "type": ["float", "null"], "doc": "y-image position of candidate [pixels]"},
-		{"name": "ra", "type": ["double", "null"], "doc": "Right Ascension of candidate; J2000 [deg]"},
-		{"name": "dec", "type": ["double", "null"], "doc": "Declination of candidate; J2000 [deg]"},
-		{"name": "magpsf", "type": ["float", "null"], "doc": "Magnitude from PSF-fit photometry [mag]"},
-		{"name": "sigmapsf", "type": ["float", "null"], "doc": "1-sigma uncertainty in magpsf [mag]"},
+		{"name": "fieldid", "type": "int", "doc": "field ID"},
+		{"name": "ra", "type": "double", "doc": "Right Ascension of candidate; J2000 [deg]"},
+		{"name": "dec", "type": "double", "doc": "Declination of candidate; J2000 [deg]"},
+		{"name": "magpsf", "type": "float", "doc": "Magnitude from PSF-fit photometry [mag]"},
+		{"name": "sigmapsf", "type": "float", "doc": "1-sigma uncertainty in magpsf [mag]"},
 		{"name": "fwhm", "type": ["float", "null"], "doc": "Full Width Half Max assuming a Gaussian core, from SExtractor [pixels]"},
 		{"name": "scorr", "type": ["double", "null"], "doc": "Peak-pixel signal-to-noise ratio in point source matched-filtered detection image"}
                 ]
diff --git a/mirar/pipelines/winter/generator.py b/mirar/pipelines/winter/generator.py
index c03eb04f9..6e8c33cd9 100644
--- a/mirar/pipelines/winter/generator.py
+++ b/mirar/pipelines/winter/generator.py
@@ -10,6 +10,8 @@
 from mirar.catalog import Gaia2Mass
 from mirar.catalog.vizier import PS1
 from mirar.data import Image
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.transactions import select_from_table
 from mirar.paths import SATURATE_KEY, get_output_dir
 from mirar.pipelines.winter.config import (
     psfex_path,
@@ -255,22 +257,22 @@ def winter_reference_generator(image: Image):
     subdetid = int(image[SUB_ID_KEY])
     logger.debug(f"Fieldid: {fieldid}, subdetid: {subdetid}")
 
-    db_results = RefStack.sql_model().select_query(
-        select_keys=["savepath"],
-        compare_keys=["fieldid", SUB_ID_KEY.lower()],
-        compare_values=[fieldid, subdetid],
-        comparators=["__eq__", "__eq__"],
+    constraints = DBQueryConstraints(
+        columns=["fieldid", SUB_ID_KEY.lower()],
+        accepted_values=[fieldid, subdetid],
     )
 
-    ref_exists = False
-    if len(db_results) > 0:
-        savepaths = [x[0] for x in db_results]
-        if os.path.exists(savepaths[0]):
-            ref_exists = True
-            logger.debug(f"Found reference image in database: {savepaths[0]}")
+    db_results = select_from_table(
+        db_constraints=constraints,
+        sql_table=RefStack.sql_model,
+        output_columns=["savepath"],
+    )
 
-    if ref_exists:
-        return RefFromPath(path=savepaths[0], filter_name=filtername)
+    if len(db_results) > 0:
+        savepath = db_results["savepath"].iloc[0]
+        if os.path.exists(savepath):
+            logger.debug(f"Found reference image in database: {savepath}")
+            return RefFromPath(path=savepath, filter_name=filtername)
 
     ukirt_query = UKIRTOnlineQuery(
         num_query_points=9,
diff --git a/mirar/pipelines/winter/ingest_references.py b/mirar/pipelines/winter/ingest_references.py
index 2c306d239..37c071406 100644
--- a/mirar/pipelines/winter/ingest_references.py
+++ b/mirar/pipelines/winter/ingest_references.py
@@ -9,9 +9,9 @@
 
 from astropy.io import fits
 
+from mirar.database.user import PostgresUser
 from mirar.paths import LATEST_SAVE_KEY, base_output_dir, get_output_dir
 from mirar.pipelines.winter.models import RefComponent
-from mirar.processors.sqldatabase.postgres import PostgresUser
 
 logger = logging.getLogger(__name__)
 
diff --git a/mirar/pipelines/winter/load_winter_image.py b/mirar/pipelines/winter/load_winter_image.py
index 41ddf90f7..95c7161c9 100644
--- a/mirar/pipelines/winter/load_winter_image.py
+++ b/mirar/pipelines/winter/load_winter_image.py
@@ -148,6 +148,12 @@ def clean_header(header: fits.Header) -> fits.Header:
     if header["PROGNAME"] == "":
         header["PROGNAME"] = default_program.progname
 
+    if len(header["PROGNAME"]) != 8:
+        logger.warning(
+            f"PROGNAME {header['PROGNAME']} is not 8 characters long. Replacing with default."
+        )
+        header["PROGNAME"] = default_program.progname
+
     with warnings.catch_warnings():
         warnings.simplefilter("ignore", AstropyWarning)
         header["NIGHTDATE"] = date_t.to_datetime().strftime("%Y-%m-%d")
diff --git a/mirar/pipelines/winter/models/__init__.py b/mirar/pipelines/winter/models/__init__.py
index 705d00ee5..b97d7b378 100644
--- a/mirar/pipelines/winter/models/__init__.py
+++ b/mirar/pipelines/winter/models/__init__.py
@@ -1,15 +1,23 @@
 """
 Models for database and pydantic dataclass models
 """
-from typing import Union
+import logging
 
-from sqlalchemy.orm import DeclarativeBase
-
-from mirar.pipelines.winter.constants import NXSPLIT, NYSPLIT
+from mirar.database.base_table import BaseTable
+from mirar.database.credentials import DB_USER
+from mirar.database.q3c import create_q3c_extension
+from mirar.database.setup import setup_database
 from mirar.pipelines.winter.models._astrometry_stats import (
     AstrometryStat,
     AstrometryStatsTable,
 )
+from mirar.pipelines.winter.models._candidates import (
+    CANDIDATE_PREFIX,
+    NAME_START,
+    Candidate,
+    CandidatesTable,
+)
+from mirar.pipelines.winter.models._diff import Diff, DiffsTable
 from mirar.pipelines.winter.models._exposures import Exposure, ExposuresTable
 from mirar.pipelines.winter.models._fields import (
     DEFAULT_FIELD,
@@ -39,7 +47,7 @@
     default_program,
     populate_programs,
 )
-from mirar.pipelines.winter.models._raw import Raw, RawTable
+from mirar.pipelines.winter.models._raw import Raw, RawsTable
 from mirar.pipelines.winter.models._ref_components import (
     RefComponent,
     RefComponentsTable,
@@ -53,47 +61,31 @@
     populate_subdets,
 )
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.database.postgres import (
-    ADMIN_PASSWORD,
-    ADMIN_USER,
-    DB_PASSWORD,
-    DB_USER,
-    PostgresAdmin,
-)
-from mirar.processors.sqldatabase.base_model import BaseTable
-from mirar.utils.sql import get_engine
 
+logger = logging.getLogger(__name__)
 
-def setup_database(base: Union[DeclarativeBase, BaseTable]):
-    """
-    Function to setup database
-    Args:
-        base:
-    Returns:
-    """
-    if DB_USER is not None:
-        db_name = base.db_name
-        admin_engine = get_engine(
-            db_name=db_name, db_user=ADMIN_USER, db_password=ADMIN_PASSWORD
-        )
-
-        pg_admin = PostgresAdmin()
 
-        if not pg_admin.check_if_db_exists(db_name=db_name):
-            pg_admin.create_db(db_name=db_name)
-
-        base.metadata.create_all(
-            admin_engine
-        )  # extensions need to be created as a superuser
-
-        if not pg_admin.check_if_user_exists(user_name=DB_USER):
-            pg_admin.create_new_user(new_db_user=DB_USER, new_password=DB_PASSWORD)
+def set_up_q3c(db_name: str, db_table: BaseTable):
+    """
+    Function to setup q3c extension for a given table in db
 
-        pg_admin.grant_privileges(db_name=db_name, db_user=DB_USER)
+    :param db_name: Name of database
+    :param db_table: Table to setup q3c extension for
+    :return:
+    """
+    create_q3c_extension(
+        db_name=db_name,
+        table_name=db_table.__tablename__,
+        ra_column_name=db_table.ra_column_name,
+        dec_column_name=db_table.dec_column_name,
+    )
 
 
 if DB_USER is not None:
-    setup_database(base=WinterBase)
+    setup_database(db_base=WinterBase)
+
+    for table in [ExposuresTable, CandidatesTable]:
+        set_up_q3c(db_name=WinterBase.db_name, db_table=table)
 
     populate_fields()
     populate_itid()
diff --git a/mirar/pipelines/winter/models/_astrometry_stats.py b/mirar/pipelines/winter/models/_astrometry_stats.py
index c39845666..6a5b0aaf3 100644
--- a/mirar/pipelines/winter/models/_astrometry_stats.py
+++ b/mirar/pipelines/winter/models/_astrometry_stats.py
@@ -5,15 +5,12 @@
 from typing import ClassVar
 
 from pydantic import Field
-from sqlalchemy import Column, Float, ForeignKey  # event,
+from sqlalchemy import Column, Float, ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
-from mirar.pipelines.winter.models._raw import RawTable
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.pipelines.winter.models._raw import RawsTable
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB, dec_field, ra_field
-
-# from mirar.utils.sql import create_q3c_extension
-
 
 logger = logging.getLogger(__name__)
 
@@ -26,8 +23,8 @@ class AstrometryStatsTable(WinterBase):  # pylint: disable=too-few-public-method
     __tablename__ = "astrometry_stats"
     __table_args__ = {"extend_existing": True}
 
-    rawid = mapped_column(ForeignKey("raw.rawid"), primary_key=True)
-    astrom_raw_ids: Mapped["RawTable"] = relationship(back_populates="astrometry")
+    rawid = mapped_column(ForeignKey("raws.rawid"), primary_key=True, unique=True)
+    astrom_raw_ids: Mapped["RawsTable"] = relationship(back_populates="astrometry")
     crval1 = Column(Float)
     crval2 = Column(Float)
     crpix1 = Column(Float)
@@ -50,16 +47,6 @@ class AstrometryStatsTable(WinterBase):  # pylint: disable=too-few-public-method
     dec_column_name = "crval2"
 
 
-# @event.listens_for(target=RawTable.__table__, identifier="after_create")
-# def raw_q3c(tbl, conn, *args, **kw):
-#     create_q3c_extension(
-#         conn=conn,
-#         __tablename__=RawTable.__tablename__,
-#         ra_column_name=RawTable.ra_column_name,
-#         dec_column_name=RawTable.dec_column_name,
-#     )
-
-
 default_unknown_field = Field(default=-999)
 
 
@@ -95,4 +82,4 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.rawid, keys="rawid")
+        return self._exists(values=self.rawid, keys="rawid")
diff --git a/mirar/pipelines/winter/models/_candidates.py b/mirar/pipelines/winter/models/_candidates.py
new file mode 100644
index 000000000..e4c791bbf
--- /dev/null
+++ b/mirar/pipelines/winter/models/_candidates.py
@@ -0,0 +1,332 @@
+"""
+Models for the 'candidates' table
+"""
+import logging
+from typing import ClassVar
+
+from pydantic import Field, validator
+from sqlalchemy import (
+    VARCHAR,
+    BigInteger,
+    Boolean,
+    Column,
+    Float,
+    ForeignKey,
+    Integer,
+    Sequence,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.pipelines.winter.models._diff import Diff
+from mirar.pipelines.winter.models._fields import fieldid_field
+from mirar.pipelines.winter.models._filters import fid_field
+from mirar.pipelines.winter.models.base_model import WinterBase
+
+logger = logging.getLogger(__name__)
+
+CANDIDATE_PREFIX = "WNTR"
+NAME_START = "aaaaa"
+
+MIN_NAME_LENGTH = len(CANDIDATE_PREFIX) + len(NAME_START) + 2
+
+
+class CandidatesTable(WinterBase):  # pylint: disable=too-few-public-methods
+    """
+    Raw table in database
+    """
+
+    __tablename__ = "candidates"
+    __table_args__ = {"extend_existing": True}
+
+    # extra avro_path, diff img foreign key etc
+
+    # Core fields
+    candid = Column(
+        BigInteger,
+        Sequence(name="candidates_candid_seq", start=1, increment=1),
+        unique=True,
+        autoincrement=True,
+        primary_key=True,
+    )
+    objectid = Column(VARCHAR(40), nullable=False, unique=False)
+    deprecated = Column(Boolean, nullable=False, default=False)
+    jd = Column(Float, nullable=False)
+
+    # Image properties
+
+    diffid: Mapped[int] = mapped_column(ForeignKey("diffs.diffid"))  # FIXME
+
+    fid: Mapped[int] = mapped_column(ForeignKey("filters.fid"))
+    exptime = Column(Float, nullable=False)
+
+    progname: Mapped[str] = mapped_column(ForeignKey("programs.progname"))
+
+    isdiffpos = Column(Boolean, nullable=False)
+
+    fieldid: Mapped[int] = mapped_column(ForeignKey("fields.fieldid"))
+
+    # Positional properties
+
+    ra = Column(Float)
+    dec = Column(Float)
+    ra_column_name = "ra"
+    dec_column_name = "dec"
+
+    # Zero point properties
+
+    magzpsci = Column(Float, nullable=True)
+    magzpsciunc = Column(Float, nullable=True)
+    magzpscirms = Column(Float, nullable=True)
+
+    # Photometry properties
+
+    diffmaglim = Column(Float, nullable=True)
+
+    magpsf = Column(Float, nullable=False)
+    sigmapsf = Column(Float, nullable=False)
+    chipsf = Column(Float, nullable=True)
+
+    magap = Column(Float, nullable=True)
+    sigmagap = Column(Float, nullable=True)
+
+    magapbig = Column(Float, nullable=True)
+    sigmagapbig = Column(Float, nullable=True)
+
+    magdiff = Column(Float, nullable=True)
+    magfromlim = Column(Float, nullable=True)
+
+    # Diagnostic properties
+
+    distnr = Column(Float, nullable=True)
+    magnr = Column(Float, nullable=True)
+    sigmagnr = Column(Float, nullable=True)
+
+    xpos = Column(Float, nullable=True)
+    ypos = Column(Float, nullable=True)
+
+    sky = Column(Float, nullable=True)
+    fwhm = Column(Float, nullable=False)
+    mindtoedge = Column(Float, nullable=True)
+    seeratio = Column(Float, nullable=True)
+    aimage = Column(Float, nullable=False)
+    bimage = Column(Float, nullable=False)
+    aimagerat = Column(Float, nullable=False)
+    bimagerat = Column(Float, nullable=False)
+    elong = Column(Float, nullable=False)
+    nneg = Column(Integer, nullable=True)
+    nbad = Column(Integer, nullable=True)
+    sumrat = Column(Float, nullable=True)
+
+    # Diff/ZOGY properties
+
+    dsnrms = Column(Float, nullable=True)
+    ssnrms = Column(Float, nullable=True)
+    dsdiff = Column(Float, nullable=True)
+
+    scorr = Column(Float, nullable=False)
+
+    # Real/bogus properties
+
+    rb = Column(Float, nullable=True)
+    rbversion = Column(Float, nullable=True)
+
+    # Solar system properties
+
+    ssdistnr = Column(Float, nullable=True)
+    ssmagnr = Column(Float, nullable=True)
+    ssnamenr = Column(VARCHAR(40), nullable=True)
+
+    # ToO
+
+    tooflag = Column(Boolean, nullable=False)
+
+    # Cross-match properties
+
+    # Ps1 properties
+
+    psobjectid1 = Column(VARCHAR(40), nullable=True)
+    sgmag1 = Column(Float, nullable=True)
+    srmag1 = Column(Float, nullable=True)
+    simag1 = Column(Float, nullable=True)
+    szmag1 = Column(Float, nullable=True)
+    distpsnr1 = Column(Float, nullable=True)
+    sgscore1 = Column(Float, nullable=True)
+
+    psobjectid2 = Column(VARCHAR(40), nullable=True)
+    sgmag2 = Column(Float, nullable=True)
+    srmag2 = Column(Float, nullable=True)
+    simag2 = Column(Float, nullable=True)
+    szmag2 = Column(Float, nullable=True)
+    distpsnr2 = Column(Float, nullable=True)
+    sgscore2 = Column(Float, nullable=True)
+
+    psobjectid3 = Column(VARCHAR(40), nullable=True)
+    sgmag3 = Column(Float, nullable=True)
+    srmag3 = Column(Float, nullable=True)
+    simag3 = Column(Float, nullable=True)
+    szmag3 = Column(Float, nullable=True)
+    distpsnr3 = Column(Float, nullable=True)
+    sgscore3 = Column(Float, nullable=True)
+
+    # 2Mass properties
+
+    tmjmag1 = Column(Float, nullable=True)
+    tmhmag1 = Column(Float, nullable=True)
+    tmkmag1 = Column(Float, nullable=True)
+    tmobjectid1 = Column(VARCHAR(40), nullable=True)
+
+    tmjmag2 = Column(Float, nullable=True)
+    tmhmag2 = Column(Float, nullable=True)
+    tmkmag2 = Column(Float, nullable=True)
+    tmobjectid2 = Column(VARCHAR(40), nullable=True)
+
+    tmjmag3 = Column(Float, nullable=True)
+    tmhmag3 = Column(Float, nullable=True)
+    tmkmag3 = Column(Float, nullable=True)
+    tmobjectid3 = Column(VARCHAR(40), nullable=True)
+
+    # Gaia properties
+
+    neargaia = Column(Float, nullable=True)
+    neargaiabright = Column(Float, nullable=True)
+    maggaia = Column(Float, nullable=True)
+    maggaiabright = Column(Float, nullable=True)
+
+
+class Candidate(BaseDB):
+    """
+    A pydantic model for a candidate database entry
+    """
+
+    sql_model: ClassVar = CandidatesTable
+
+    objectid: str = Field(min_length=MIN_NAME_LENGTH)
+    deprecated: bool = Field(default=False)
+
+    jd: float = Field(ge=0)
+
+    diffid: int | None = Field(ge=0, default=None)
+
+    fid: int = fid_field
+    exptime: float = Field(ge=0)
+
+    progname: str = Field(min_length=8, max_length=8, example="2020A000")
+
+    isdiffpos: bool = Field(default=True)
+
+    fieldid: int = fieldid_field
+
+    ra: float = ra_field
+    dec: float = dec_field
+
+    magzpsci: float | None = Field(default=None)
+    magzpsciunc: float | None = Field(ge=0, default=None)
+    magzpscirms: float | None = Field(ge=0, default=None)
+
+    diffmaglim: float | None = Field(default=None)
+
+    magpsf: float = Field()
+    sigmapsf: float = Field(ge=0)
+    chipsf: float | None = Field(ge=0, default=None)
+
+    magap: float | None = Field(default=None)
+    sigmagap: float | None = Field(ge=0, default=None)
+
+    magapbig: float | None = Field(default=None)
+    sigmagapbig: float | None = Field(ge=0, default=None)
+
+    magdiff: float | None = Field(default=None)
+    magfromlim: float | None = Field(default=None)
+
+    distnr: float | None = Field(ge=0, default=None)
+    magnr: float | None = Field(ge=0, default=None)
+    sigmagnr: float | None = Field(ge=0, default=None)
+
+    xpos: float | None = Field(ge=0, default=None)
+    ypos: float | None = Field(ge=0, default=None)
+
+    sky: float | None = Field(ge=0, default=None)
+    fwhm: float = Field(ge=0)
+    mindtoedge: float | None = Field(ge=0, default=None)
+    seeratio: float | None = Field(ge=0, default=None)
+    aimage: float = Field(ge=0)
+    bimage: float = Field(ge=0)
+    aimagerat: float = Field(ge=0)
+    bimagerat: float = Field(ge=0)
+    elong: float = Field(ge=0)
+    nneg: int | None = Field(ge=0, default=None)
+    nbad: int | None = Field(ge=0, default=None)
+    sumrat: float | None = Field(ge=0, default=None)
+
+    dsnrms: float | None = Field(ge=0, default=None)
+    ssnrms: float | None = Field(ge=0, default=None)
+    dsdiff: float | None = Field(ge=0, default=None)
+
+    scorr: float = Field(ge=0)
+
+    rb: float | None = Field(ge=0, default=None)
+    rbversion: float | None = Field(ge=0, default=None)
+
+    ssdistnr: float | None = Field(ge=0, default=None)
+    ssmagnr: float | None = Field(ge=0, default=None)
+    ssnamenr: str | None = Field(min_length=MIN_NAME_LENGTH, default=None)
+
+    tooflag: bool = Field(default=False)
+
+    psobjectid1: str | None = Field(min_length=MIN_NAME_LENGTH, default=None)
+    psobjectid2: str | None = Field(min_length=MIN_NAME_LENGTH, default=None)
+    psobjectid3: str | None = Field(min_length=MIN_NAME_LENGTH, default=None)
+
+    sgmag1: float | None = Field(default=None)
+    srmag1: float | None = Field(default=None)
+    simag1: float | None = Field(default=None)
+    szmag1: float | None = Field(default=None)
+    sgscore1: float | None = Field(ge=0, default=None)
+    distpsnr1: float | None = Field(ge=0, default=None)
+
+    sgmag2: float | None = Field(default=None)
+    srmag2: float | None = Field(default=None)
+    simag2: float | None = Field(default=None)
+    szmag2: float | None = Field(default=None)
+    sgscore2: float | None = Field(ge=0, default=None)
+    distpsnr2: float | None = Field(ge=0, default=None)
+
+    sgmag3: float | None = Field(default=None)
+    srmag3: float | None = Field(default=None)
+    simag3: float | None = Field(default=None)
+    szmag3: float | None = Field(default=None)
+    sgscore3: float | None = Field(ge=0, default=None)
+    distpsnr3: float | None = Field(ge=0, default=None)
+
+    tmjmag1: float | None = Field(default=None)
+    tmhmag1: float | None = Field(default=None)
+    tmkmag1: float | None = Field(default=None)
+    tmobjectid1: str | None = Field(default=None)
+
+    tmjmag2: float | None = Field(default=None)
+    tmhmag2: float | None = Field(default=None)
+    tmkmag2: float | None = Field(default=None)
+    tmobjectid2: str | None = Field(default=None)
+
+    tmjmag3: float | None = Field(default=None)
+    tmhmag3: float | None = Field(default=None)
+    tmkmag3: float | None = Field(default=None)
+    tmobjectid3: str | None = Field(default=None)
+
+    neargaia: float | None = Field(ge=0, default=None)
+    neargaiabright: float | None = Field(ge=0, default=None)
+    maggaia: float | None = Field(default=None)
+    maggaiabright: float | None = Field(default=None)
+
+    @validator("diffid")
+    @classmethod
+    def validate_diffid(cls, field_value: int):
+        """
+        Ensure that expid exists in exposures table
+
+        :param field_value: field value
+        :return: field value
+        """
+        assert Diff._exists(keys="diffid", values=field_value)
+        return field_value
diff --git a/mirar/pipelines/winter/models/_diff.py b/mirar/pipelines/winter/models/_diff.py
new file mode 100644
index 000000000..9e2ec8a25
--- /dev/null
+++ b/mirar/pipelines/winter/models/_diff.py
@@ -0,0 +1,79 @@
+"""
+Models for the 'diff' table
+"""
+import os
+from typing import ClassVar
+
+from pydantic import Field, validator
+from sqlalchemy import VARCHAR, Column, ForeignKey, Integer, Sequence
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from mirar.database.base_model import BaseDB
+from mirar.pipelines.winter.models._exposures import Exposure
+from mirar.pipelines.winter.models.base_model import WinterBase
+
+
+class DiffsTable(WinterBase):  # pylint: disable=too-few-public-methods
+    """
+    Diffs table in database
+    """
+
+    __tablename__ = "diffs"
+    __table_args__ = {"extend_existing": True}
+
+    diffid = Column(
+        Integer,
+        Sequence(start=1, name="diff_udiffid_seq"),
+        autoincrement=True,
+        unique=True,
+        primary_key=True,
+    )
+
+    uexpid: Mapped[int] = mapped_column(ForeignKey("exposures.uexpid"))
+    exposure_ids: Mapped["ExposuresTable"] = relationship(back_populates="diff")
+
+    rawid: Mapped[int] = mapped_column(ForeignKey("raws.rawid"))
+    raw_ids: Mapped["RawsTable"] = relationship(back_populates="diff")
+
+    stackid: Mapped[int] = mapped_column(ForeignKey("stacks.stackid"))
+    stack_id: Mapped["StacksTable"] = relationship(back_populates="diff")
+
+    savepath = Column(VARCHAR(255), unique=True)
+
+
+class Diff(BaseDB):
+    """
+    A pydantic model for a diff database entry
+    """
+
+    sql_model: ClassVar = DiffsTable
+
+    diffid: int = Field(ge=0)
+    uexpid: int = Field(ge=0)
+    qid: int = Field(ge=0)
+    savepath: str = Field(min_length=1)
+    procstatus: int = Field(ge=0, default=0)
+
+    @validator("savepath")
+    @classmethod
+    def validate_savepath(cls, field_value: str):
+        """
+        Ensure that path exists
+
+        :param field_value: field value
+        :return: field value
+        """
+        assert os.path.exists(field_value)
+        return field_value
+
+    @validator("uexpid")
+    @classmethod
+    def validate_expid(cls, field_value: int):
+        """
+        Ensure that expid exists in exposures table
+
+        :param field_value: field value
+        :return: field value
+        """
+        assert Exposure._exists(keys="uexpid", values=field_value)
+        return field_value
diff --git a/mirar/pipelines/winter/models/_exposures.py b/mirar/pipelines/winter/models/_exposures.py
index 8811ea7df..72f5f3f8f 100644
--- a/mirar/pipelines/winter/models/_exposures.py
+++ b/mirar/pipelines/winter/models/_exposures.py
@@ -5,11 +5,12 @@
 from datetime import date, datetime
 from typing import ClassVar
 
+import pandas as pd
 from pydantic import Field
 from sqlalchemy import (  # event,
+    BigInteger,
     Column,
     DateTime,
-    Double,
     Float,
     ForeignKey,
     Integer,
@@ -17,22 +18,15 @@
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+from mirar.database.base_model import BaseDB, alt_field, az_field, dec_field, ra_field
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.transactions import select_from_table
 from mirar.pipelines.winter.models._fields import FieldsTable, fieldid_field
 from mirar.pipelines.winter.models._filters import FiltersTable, fid_field
 from mirar.pipelines.winter.models._img_type import ImgTypesTable
 from mirar.pipelines.winter.models._nights import Night, NightsTable
-from mirar.pipelines.winter.models._programs import ProgramsTable, default_program
+from mirar.pipelines.winter.models._programs import Program, default_program
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import (
-    BaseDB,
-    alt_field,
-    az_field,
-    dec_field,
-    ra_field,
-)
-
-# from mirar.utils.sql import create_q3c_extension
-
 
 logger = logging.getLogger(__name__)
 
@@ -50,8 +44,9 @@ class ExposuresTable(WinterBase):  # pylint: disable=too-few-public-methods
         Sequence(name="exposures_uexpid_seq", start=1, increment=1),
         unique=True,
         autoincrement=True,
+        primary_key=True,
     )
-    expid = Column(Double, primary_key=True, unique=True, autoincrement=False)
+    expid = Column(BigInteger, primary_key=False, unique=True, autoincrement=False)
     # Deterministic ID of exposure
 
     fid: Mapped[int] = mapped_column(ForeignKey("filters.fid"))
@@ -71,18 +66,18 @@ class ExposuresTable(WinterBase):  # pylint: disable=too-few-public-methods
 
     utctime = Column(DateTime(timezone=True))
 
-    ExpTime = Column(Float, nullable=False)
-    expMJD = Column(Float, nullable=False)
+    exptime = Column(Float, nullable=False)
+    expmjd = Column(Float, nullable=False)
     airmass = Column(Float)
     tempture = Column(Float, default=-999)
     windspd = Column(Float, default=-999)
-    Dewpoint = Column(Float, default=-999)
-    Humidity = Column(Float, default=-999)
-    Pressure = Column(Float, default=-999)
+    dewpoint = Column(Float, default=-999)
+    humidity = Column(Float, default=-999)
+    pressure = Column(Float, default=-999)
 
-    Moonaz = Column(Float, default=-999)
-    Moonalt = Column(Float, default=-999)
-    Sunalt = Column(Float, default=-999)
+    moonaz = Column(Float, default=-999)
+    moonalt = Column(Float, default=-999)
+    sunalt = Column(Float, default=-999)
 
     ra = Column(Float)
     dec = Column(Float)
@@ -92,17 +87,8 @@ class ExposuresTable(WinterBase):  # pylint: disable=too-few-public-methods
     ra_column_name = "ra"
     dec_column_name = "dec"
 
-    raw: Mapped["RawTable"] = relationship(back_populates="exposure_ids")
-
-
-# @event.listens_for(target=RawTable.__table__, identifier="after_create")
-# def raw_q3c(tbl, conn, *args, **kw):
-#     create_q3c_extension(
-#         conn=conn,
-#         __tablename__=RawTable.__tablename__,
-#         ra_column_name=RawTable.ra_column_name,
-#         dec_column_name=RawTable.dec_column_name,
-#     )
+    raw: Mapped["RawsTable"] = relationship(back_populates="exposure_ids")
+    diff: Mapped["DiffsTable"] = relationship(back_populates="exposure_ids")
 
 
 default_unknown_field = Field(default=-999)
@@ -123,25 +109,25 @@ class Exposure(BaseDB):
     progname: str = Field(min_length=1)
 
     utctime: datetime = Field()
-    ExpTime: float = Field(ge=0)
-    expMJD: float = Field(ge=59000)
+    exptime: float = Field(ge=0)
+    expmjd: float = Field(ge=59000)
 
     tempture: float = default_unknown_field
     windspd: float = default_unknown_field
-    Dewpoint: float = default_unknown_field
-    Humidity: float = default_unknown_field
-    Pressure: float = default_unknown_field
+    dewpoint: float = default_unknown_field
+    humidity: float = default_unknown_field
+    pressure: float = default_unknown_field
 
-    Moonaz: float = default_unknown_field
-    Moonalt: float = default_unknown_field
-    Sunalt: float = default_unknown_field
+    moonaz: float = default_unknown_field
+    moonalt: float = default_unknown_field
+    sunalt: float = default_unknown_field
 
     ra: float = ra_field
     dec: float = dec_field
     altitude: float = alt_field
     azimuth: float = az_field
 
-    def insert_entry(self, returning_key_names=None) -> tuple:
+    def insert_entry(self, returning_key_names=None) -> pd.DataFrame:
         """
         Insert the pydantic-ified data into the corresponding sql database
 
@@ -152,15 +138,18 @@ def insert_entry(self, returning_key_names=None) -> tuple:
         if not night.exists():
             night.insert_entry()
 
-        if not ProgramsTable().exists(values=self.progname, keys="progname"):
-            default_progname = ProgramsTable().select_query(
-                select_keys="progname",
-                compare_values=[default_program.progname],
-                compare_keys=["progname"],
-            )[0][0]
-            self.progname = default_progname
+        prog_match = select_from_table(
+            DBQueryConstraints(columns="progname", accepted_values=self.progname),
+            sql_table=Program.sql_model,
+        )
+        if prog_match.empty:
+            logger.debug(
+                f"Program {self.progname} not found in database. "
+                f"Using default program {default_program.progname}"
+            )
+            self.progname = default_program.progname
 
-        return self._insert_entry()
+        return self._insert_entry(returning_key_names=returning_key_names)
 
     def exists(self) -> bool:
         """
@@ -168,4 +157,4 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.expid, keys="expid")
+        return self._exists(values=self.expid, keys="expid")
diff --git a/mirar/pipelines/winter/models/_fields.py b/mirar/pipelines/winter/models/_fields.py
index f993a252d..a53a2b743 100644
--- a/mirar/pipelines/winter/models/_fields.py
+++ b/mirar/pipelines/winter/models/_fields.py
@@ -10,9 +10,11 @@
 from tqdm import tqdm
 from wintertoo.data import winter_fields
 
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.database.engine import get_engine
+from mirar.database.transactions import check_table_exists
+from mirar.database.transactions.select import is_populated
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB, _exists, dec_field, ra_field
-from mirar.utils.sql import get_engine
 
 DEFAULT_FIELD = 999999999
 
@@ -59,7 +61,8 @@ def populate_fields():
     """
 
     engine = get_engine(db_name=FieldsTable.db_name)
-    if not _exists(Select(FieldsTable), engine=engine):
+
+    if not is_populated(FieldsTable):
         chunk = 10000
 
         winter_fields["fieldid"] = winter_fields["ID"]
diff --git a/mirar/pipelines/winter/models/_filters.py b/mirar/pipelines/winter/models/_filters.py
index 478dc7694..75f56ccc7 100644
--- a/mirar/pipelines/winter/models/_filters.py
+++ b/mirar/pipelines/winter/models/_filters.py
@@ -7,9 +7,9 @@
 from sqlalchemy import VARCHAR, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB
 from mirar.pipelines.winter.constants import winter_filters_map
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB
 
 fid_field: int = Field(ge=0)
 
@@ -42,7 +42,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.fid, keys="fid")
+        return self._exists(values=self.fid, keys="fid")
 
     @validator("fid")
     @classmethod
diff --git a/mirar/pipelines/winter/models/_img_type.py b/mirar/pipelines/winter/models/_img_type.py
index bbdcf8e3c..011f51134 100644
--- a/mirar/pipelines/winter/models/_img_type.py
+++ b/mirar/pipelines/winter/models/_img_type.py
@@ -7,8 +7,8 @@
 from sqlalchemy import VARCHAR, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB
 
 ALL_ITID = ["SCIENCE", "CAL", "FOCUS", "POINTING", "NULL"]
 itid_dict = {}
@@ -46,7 +46,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.itid, keys="itid")
+        return self._exists(values=self.itid, keys="itid")
 
 
 def populate_itid():
diff --git a/mirar/pipelines/winter/models/_nights.py b/mirar/pipelines/winter/models/_nights.py
index 1a3d46090..837a61200 100644
--- a/mirar/pipelines/winter/models/_nights.py
+++ b/mirar/pipelines/winter/models/_nights.py
@@ -7,8 +7,8 @@
 from sqlalchemy import DATE, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB, date_field
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB, date_field
 
 
 class NightsTable(WinterBase):  # pylint: disable=too-few-public-methods
@@ -38,7 +38,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.nightdate, keys="nightdate")
+        return self._exists(values=self.nightdate, keys="nightdate")
 
     #
     # def increment_raw(self):
diff --git a/mirar/pipelines/winter/models/_programs.py b/mirar/pipelines/winter/models/_programs.py
index bf8b310f9..ed29ba2ac 100644
--- a/mirar/pipelines/winter/models/_programs.py
+++ b/mirar/pipelines/winter/models/_programs.py
@@ -8,8 +8,8 @@
 from sqlalchemy import CHAR, DATE, REAL, VARCHAR, Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB, date_field
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB, date_field
 from mirar.utils.security import generate_key
 
 LEN_PROG_KEY = 60
@@ -40,12 +40,15 @@ class ProgramsTable(WinterBase):  # pylint: disable=too-few-public-methods
     exposures: Mapped["ExposuresTable"] = relationship(back_populates="program_name")
 
 
+prog_field: str = Field(min_length=8, max_length=8, example="2020A000")
+
+
 class ProgramCredentials(BaseModel):
     """
     Program credentials to access a program
     """
 
-    progname: str = Field(min_length=8, max_length=8, example="2020A000")
+    progname: str = prog_field
     prog_key: str = Field(
         min_length=LEN_PROG_KEY,
         max_length=LEN_PROG_KEY,
@@ -105,7 +108,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(values=self.progname, keys="progname")
+        return self._exists(values=self.progname, keys="progname")
 
 
 default_program = Program(
diff --git a/mirar/pipelines/winter/models/_raw.py b/mirar/pipelines/winter/models/_raw.py
index cfe6d1efa..b69532d8f 100644
--- a/mirar/pipelines/winter/models/_raw.py
+++ b/mirar/pipelines/winter/models/_raw.py
@@ -5,28 +5,29 @@
 from typing import ClassVar
 
 from pydantic import Field, validator
-from sqlalchemy import VARCHAR, Column, Double, Float, ForeignKey, Integer, Sequence
-
-# event,
+from sqlalchemy import VARCHAR, BigInteger, Column, Float, ForeignKey, Integer, Sequence
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+from mirar.database.base_model import BaseDB
 from mirar.pipelines.winter.models._exposures import Exposure
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB
 
 
-class RawTable(WinterBase):  # pylint: disable=too-few-public-methods
+class RawsTable(WinterBase):  # pylint: disable=too-few-public-methods
     """
     Raw table in database
     """
 
-    __tablename__ = "raw"
+    __tablename__ = "raws"
     __table_args__ = {"extend_existing": True}
 
     urawid = Column(
-        Integer, Sequence(start=1, name="raw_urawid_seq"), autoincrement=True
+        Integer,
+        Sequence(start=1, name="raw_urawid_seq"),
+        autoincrement=True,
+        primary_key=True,
     )
-    rawid = Column(Double, primary_key=True, unique=True, autoincrement=False)
+    rawid = Column(BigInteger, primary_key=False, unique=True, autoincrement=False)
 
     uexpid: Mapped[int] = mapped_column(ForeignKey("exposures.uexpid"))
     exposure_ids: Mapped["ExposuresTable"] = relationship(back_populates="raw")
@@ -37,14 +38,13 @@ class RawTable(WinterBase):  # pylint: disable=too-few-public-methods
 
     savepath = Column(VARCHAR(255), unique=True)
 
-    # procstatus = Column(Integer, default=0)
     ustackid: Mapped[int] = mapped_column(ForeignKey("stacks.ustackid"), nullable=True)
     stacks: Mapped["StacksTable"] = relationship(back_populates="raw")
 
-    # proc: Mapped["ProcTable"] = relationship(back_populates="raw_ids")
     astrometry: Mapped["AstrometryStatsTable"] = relationship(
         back_populates="astrom_raw_ids"
     )
+    diff: Mapped["DiffsTable"] = relationship(back_populates="raw_ids")
 
 
 class Raw(BaseDB):
@@ -52,7 +52,7 @@ class Raw(BaseDB):
     A pydantic model for a raw database entry
     """
 
-    sql_model: ClassVar = RawTable
+    sql_model: ClassVar = RawsTable
 
     rawid: int = Field(ge=0)
     uexpid: int = Field(ge=0)
@@ -84,5 +84,5 @@ def validate_expid(cls, field_value: int):
         Returns:
 
         """
-        assert Exposure.sql_model().exists(keys="uexpid", values=field_value)
+        assert Exposure._exists(keys="uexpid", values=field_value)
         return field_value
diff --git a/mirar/pipelines/winter/models/_ref_components.py b/mirar/pipelines/winter/models/_ref_components.py
index 4a8853e55..2097b525b 100644
--- a/mirar/pipelines/winter/models/_ref_components.py
+++ b/mirar/pipelines/winter/models/_ref_components.py
@@ -7,8 +7,8 @@
 from sqlalchemy import VARCHAR, BigInteger, Column, Float, Integer
 from sqlalchemy.orm import Mapped, relationship
 
-from mirar.pipelines.winter.models.base_model import WinterBase, dec_field, ra_field
-from mirar.processors.sqldatabase.base_model import BaseDB
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.pipelines.winter.models.base_model import WinterBase
 
 
 class RefComponentsTable(WinterBase):
@@ -71,7 +71,4 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(
-            values=[self.mfid, self.xtnsnid],
-            keys=["mfid", "xtnsnid"],
-        )
+        return self._exists(values=[self.mfid, self.xtnsnid], keys=["mfid", "xtnsnid"])
diff --git a/mirar/pipelines/winter/models/_ref_queries.py b/mirar/pipelines/winter/models/_ref_queries.py
index 43786891b..329239936 100644
--- a/mirar/pipelines/winter/models/_ref_queries.py
+++ b/mirar/pipelines/winter/models/_ref_queries.py
@@ -7,8 +7,8 @@
 from sqlalchemy import VARCHAR, Column, Float, ForeignKey, Integer
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
-from mirar.pipelines.winter.models.base_model import WinterBase, dec_field, ra_field
-from mirar.processors.sqldatabase.base_model import BaseDB
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.pipelines.winter.models.base_model import WinterBase
 
 
 class RefQueriesTable(WinterBase):
@@ -44,7 +44,7 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(
+        return self._exists(
             values=[self.qry_ra, self.qry_dec, self.qry_filt],
             keys=["qry_ra", "qry_dec", "qry_filt"],
         )
diff --git a/mirar/pipelines/winter/models/_ref_stacks.py b/mirar/pipelines/winter/models/_ref_stacks.py
index 82c335bbc..bef4f19ed 100644
--- a/mirar/pipelines/winter/models/_ref_stacks.py
+++ b/mirar/pipelines/winter/models/_ref_stacks.py
@@ -6,8 +6,8 @@
 from pydantic import Field
 from sqlalchemy import VARCHAR, BigInteger, Column, Float, Integer
 
-from mirar.pipelines.winter.models.base_model import WinterBase, dec_field, ra_field
-from mirar.processors.sqldatabase.base_model import BaseDB
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.pipelines.winter.models.base_model import WinterBase
 
 
 class RefStacksTable(WinterBase):
@@ -68,9 +68,6 @@ def exists(self) -> bool:
 
         :return: bool
         """
-        return self.sql_model().exists(
+        return self._exists(
             values=[self.ra_cent, self.dec_cent], keys=["ra_cent", "dec_cent"]
         )
-
-    def insert_entry(self, returning_keys=None):
-        return self._insert_entry(returning_key_names=returning_keys)
diff --git a/mirar/pipelines/winter/models/_stack.py b/mirar/pipelines/winter/models/_stack.py
index e00b3f4c2..662854f72 100644
--- a/mirar/pipelines/winter/models/_stack.py
+++ b/mirar/pipelines/winter/models/_stack.py
@@ -5,11 +5,11 @@
 from typing import ClassVar
 
 from pydantic import Field, validator
-from sqlalchemy import REAL, VARCHAR, Column, Double, Integer, Sequence  # event,
+from sqlalchemy import REAL, VARCHAR, BigInteger, Column, Integer, Sequence  # event,
 from sqlalchemy.orm import relationship
 
+from mirar.database.base_model import BaseDB, dec_field, ra_field
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB, dec_field, ra_field
 
 
 class StacksTable(WinterBase):  # pylint: disable=too-few-public-methods
@@ -25,10 +25,12 @@ class StacksTable(WinterBase):  # pylint: disable=too-few-public-methods
         Sequence(start=1, name="stacks_ustackid_seq"),
         autoincrement=True,
         unique=True,
+        primary_key=True,
     )
-    stackid = Column(Double, primary_key=True, autoincrement=False)
+    stackid = Column(BigInteger, primary_key=False, unique=True, autoincrement=False)
 
-    raw = relationship("RawTable", back_populates="stacks")
+    raw = relationship("RawsTable", back_populates="stacks")
+    diff = relationship("DiffsTable", back_populates="stack_id")
     # procid = Column(Double, primary_key=True, autoincrement=False)
 
     # rawid: Mapped[int] = Column(Integer, ForeignKey("raw.rawid"), primary_key=True)
diff --git a/mirar/pipelines/winter/models/_subdets.py b/mirar/pipelines/winter/models/_subdets.py
index ceee7d31b..8d891ac66 100644
--- a/mirar/pipelines/winter/models/_subdets.py
+++ b/mirar/pipelines/winter/models/_subdets.py
@@ -4,15 +4,13 @@
 from typing import ClassVar
 
 from pydantic import Field
-from sqlalchemy import Column, Integer, Select
+from sqlalchemy import Column, Integer
 from sqlalchemy.orm import Mapped, relationship
 
+from mirar.database.base_model import BaseDB
+from mirar.database.transactions import is_populated
 from mirar.pipelines.winter.constants import subdets
 from mirar.pipelines.winter.models.base_model import WinterBase
-from mirar.processors.sqldatabase.base_model import BaseDB, _exists
-from mirar.utils.sql import get_engine
-
-DEFAULT_FIELD = 999999999
 
 
 class SubdetsTable(WinterBase):  # pylint: disable=too-few-public-methods
@@ -30,7 +28,7 @@ class SubdetsTable(WinterBase):  # pylint: disable=too-few-public-methods
     ny = Column(Integer)
     nytot = Column(Integer)
 
-    raw: Mapped["RawTable"] = relationship(back_populates="subdets")
+    raw: Mapped["RawsTable"] = relationship(back_populates="subdets")
 
 
 class Subdet(BaseDB):
@@ -52,8 +50,7 @@ def populate_subdets():
     Creates entries in the database based on number of detectors and splits
 
     """
-    engine = get_engine(db_name=SubdetsTable.db_name)
-    if not _exists(Select(SubdetsTable), engine=engine):
+    if not is_populated(SubdetsTable):
         for _, row in subdets.iterrows():
             new = Subdet(
                 boardid=row["boardid"],
diff --git a/mirar/pipelines/winter/models/base_model.py b/mirar/pipelines/winter/models/base_model.py
index 84fec76ca..170837ffc 100644
--- a/mirar/pipelines/winter/models/base_model.py
+++ b/mirar/pipelines/winter/models/base_model.py
@@ -1,12 +1,9 @@
 """
 Base class for models
 """
-from datetime import date
-
-from pydantic import Field
 from sqlalchemy.orm import DeclarativeBase
 
-from mirar.processors.sqldatabase.base_model import BaseTable
+from mirar.database.base_table import BaseTable
 
 DB_NAME = "winter"
 
@@ -17,10 +14,3 @@ class WinterBase(DeclarativeBase, BaseTable):
     """
 
     db_name = DB_NAME
-
-
-ra_field: float = Field(title="RA (degrees)", ge=0.0, le=360.0)
-dec_field: float = Field(title="Dec (degrees)", ge=-90.0, le=90.0)
-alt_field: float = Field(title="Alt (degrees)", ge=0.0, le=90.0)
-az_field: float = Field(title="Az (degrees)", ge=0.0, le=360.0)
-date_field: date = Field()
diff --git a/mirar/pipelines/wirc/blocks.py b/mirar/pipelines/wirc/blocks.py
index 8bfd14f67..27587a96b 100644
--- a/mirar/pipelines/wirc/blocks.py
+++ b/mirar/pipelines/wirc/blocks.py
@@ -1,8 +1,10 @@
 """
 Module containing standard processing blocks for WIRC
 """
+# pylint: disable=duplicate-code
 from mirar.catalog.kowalski import PS1, TMASS
 from mirar.paths import (
+    CAND_NAME_KEY,
     FITS_MASK_KEY,
     LATEST_SAVE_KEY,
     OBSCLASS_KEY,
@@ -21,7 +23,6 @@
 )
 from mirar.pipelines.wirc.load_wirc_image import load_raw_wirc_image
 from mirar.pipelines.wirc.wirc_files import (
-    candidate_colnames,
     psfex_path,
     scamp_fp_path,
     sextractor_astrometry_config,
@@ -30,9 +31,13 @@
     sextractor_reference_config,
     swarp_sp_path,
     wirc_avro_schema_path,
-    wirc_candidate_schema_path,
     wirc_mask_path,
 )
+from mirar.pipelines.wirc.wirc_files.models import (
+    CANDIDATE_PREFIX,
+    NAME_START,
+    Candidate,
+)
 from mirar.processors.astromatic import Scamp, Sextractor, Swarp
 from mirar.processors.astromatic.psfex import PSFex
 from mirar.processors.astromatic.scamp.scamp import SCAMP_HEADER_KEY
@@ -43,8 +48,8 @@
 from mirar.processors.avro import IPACAvroExporter, SendToFritz
 from mirar.processors.csvlog import CSVLog
 from mirar.processors.dark import DarkCalibrator
-from mirar.processors.database.database_exporter import DatabaseDataframeExporter
-from mirar.processors.database.database_importer import DatabaseHistoryImporter
+from mirar.processors.database.database_inserter import DatabaseSourceInserter
+from mirar.processors.database.database_selector import DatabaseHistorySelector
 from mirar.processors.flat import SkyFlatCalibrator
 from mirar.processors.mask import (
     MaskAboveThreshold,
@@ -236,31 +241,21 @@
     XMatch(catalog=TMASS(num_sources=3, search_radius_arcmin=0.5)),
     XMatch(catalog=PS1(num_sources=3, search_radius_arcmin=0.5)),
     SourceWriter(output_dir_name="kowalski"),
-    DatabaseHistoryImporter(
+    DatabaseHistorySelector(
         crossmatch_radius_arcsec=2.0,
         time_field_name="jd",
         history_duration_days=500.0,
-        db_name="wirc",
-        db_table="candidates",
-        db_output_columns=candidate_colnames,
-        schema_path=wirc_candidate_schema_path,
-        q3c_bool=False,
+        db_table=Candidate,
+        db_output_columns=[CAND_NAME_KEY],
     ),
     CandidateNamer(
-        db_name="wirc",
-        db_table="candidates",
-        base_name="WIRC",
-        name_start="aaaaa",
+        db_table=Candidate,
+        base_name=CANDIDATE_PREFIX,
+        name_start=NAME_START,
         xmatch_radius_arcsec=2,
-        schema_path=wirc_candidate_schema_path,
-    ),
-    DatabaseDataframeExporter(
-        db_name="wirc",
-        db_table="candidates",
-        schema_path=wirc_candidate_schema_path,
-        duplicate_protocol="replace",
     ),
-    SourceWriter(output_dir_name="dbop"),
+    DatabaseSourceInserter(db_table=Candidate, duplicate_protocol="fail")
+    # SourceWriter(output_dir_name="dbop"),
     # EdgeCandidatesMask(edge_boundary_size=100)
     # FilterCandidates(),
 ]
diff --git a/mirar/pipelines/wirc/wirc_files/__init__.py b/mirar/pipelines/wirc/wirc_files/__init__.py
index 626f0f124..d053b0f4f 100644
--- a/mirar/pipelines/wirc/wirc_files/__init__.py
+++ b/mirar/pipelines/wirc/wirc_files/__init__.py
@@ -3,8 +3,6 @@
 """
 from pathlib import Path
 
-from mirar.processors.database.utils import get_column_names_from_schema
-
 wirc_file_dir = Path(__file__).parent.joinpath("files")
 wirc_schema_dir = Path(__file__).parent.joinpath("schema")
 
@@ -28,10 +26,6 @@
 
 swarp_sp_path = wirc_file_dir.joinpath("second_pass.swarp")
 
-wirc_candidate_schema_path = wirc_schema_dir.joinpath("candidates.sql")
-
-candidate_colnames = get_column_names_from_schema(wirc_candidate_schema_path)
-
 sextractor_reference_config = {
     "config_path": wirc_file_dir.joinpath("photomCat.sex"),
     "parameter_path": wirc_file_dir.joinpath("photom.param"),
diff --git a/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc b/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc
index 5cdd9baa1..582dcb09c 100644
--- a/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc
+++ b/mirar/pipelines/wirc/wirc_files/avro_schema/wirc.alert.avsc
@@ -6,7 +6,7 @@
     "fields": [
         {"name": "schemavsn", "type": "string", "doc": "schema version used", "default": "0.1"},
         {"name": "publisher", "type": "string", "doc": "origin of alert packet", "default":  "WIRC-mirar"},
-        {"name": "objectId", "type": "string"},
+        {"name": "objectid", "type": "string"},
         {"name": "candid", "type": "long"},
         {"name": "candidate", "type": "wirc.alert.candidate", "doc": "candidate information"},
         {"name": "prv_candidates", "type": [{
diff --git a/mirar/pipelines/wirc/wirc_files/models.py b/mirar/pipelines/wirc/wirc_files/models.py
new file mode 100644
index 000000000..1783c571a
--- /dev/null
+++ b/mirar/pipelines/wirc/wirc_files/models.py
@@ -0,0 +1,148 @@
+"""
+Models for the 'candidates' table
+"""
+#  pylint: disable=duplicate-code
+import logging
+from typing import ClassVar
+
+from pydantic import Field
+from sqlalchemy import VARCHAR, BigInteger, Column, Float, Integer, Sequence
+from sqlalchemy.orm import DeclarativeBase
+
+from mirar.database.base_model import BaseDB, dec_field, ra_field
+from mirar.database.base_table import BaseTable
+from mirar.database.credentials import DB_USER
+from mirar.database.setup import setup_database
+
+DB_NAME = "wirc"
+
+
+logger = logging.getLogger(__name__)
+
+CANDIDATE_PREFIX = "WIRC"
+NAME_START = "aaaaa"
+
+MIN_NAME_LENGTH = len(CANDIDATE_PREFIX) + len(NAME_START) + 2
+
+
+class WircBase(DeclarativeBase, BaseTable):
+    """
+    Parent class for summer database
+    """
+
+    db_name = DB_NAME
+
+
+class CandidatesTable(WircBase):  # pylint: disable=too-few-public-methods
+    """
+    Raw table in database
+    """
+
+    __tablename__ = "candidates"
+    __table_args__ = {"extend_existing": True}
+
+    # extra avro_path, diff img foreign key etc
+
+    # Core fields
+    candid = Column(
+        BigInteger,
+        Sequence(name="candidates_candid_seq", start=1, increment=1),
+        unique=True,
+        autoincrement=True,
+        primary_key=True,
+    )
+    objectid = Column(VARCHAR(40), nullable=False, unique=False)
+
+    # Positional properties
+
+    ra = Column(Float)
+    dec = Column(Float)
+    ra_column_name = "ra"
+    dec_column_name = "dec"
+
+    fwhm = Column(Float, nullable=True)
+
+    jd = Column(Float, nullable=False)
+
+    fid = Column(Integer, nullable=False)
+
+    diffimgname = Column(VARCHAR(255), nullable=True)
+    sciimgname = Column(VARCHAR(255), nullable=True)
+    refimgname = Column(VARCHAR(255), nullable=True)
+
+    magpsf = Column(Float, nullable=False)
+    sigmapsf = Column(Float, nullable=False)
+    chipsf = Column(Float, nullable=True)
+
+    aimage = Column(Float, nullable=False)
+    bimage = Column(Float, nullable=False)
+    aimagerat = Column(Float, nullable=False)
+    bimagerat = Column(Float, nullable=False)
+    elong = Column(Float, nullable=False)
+
+    scorr = Column(Float, nullable=False)
+
+    xpos = Column(Float, nullable=True)
+    ypos = Column(Float, nullable=True)
+
+    # Zero point properties
+
+    magzpsci = Column(Float, nullable=True)
+    magzpsciunc = Column(Float, nullable=True)
+    magzpscirms = Column(Float, nullable=True)
+
+    tmjmag1 = Column(Float, nullable=True)
+    tmhmag1 = Column(Float, nullable=True)
+    tmkmag1 = Column(Float, nullable=True)
+    tmobjectid1 = Column(VARCHAR(25), nullable=True)
+
+
+class Candidate(BaseDB):
+    """
+    A pydantic model for a candidate database entry
+    """
+
+    sql_model: ClassVar = CandidatesTable
+
+    objectid: str = Field(min_length=MIN_NAME_LENGTH)
+
+    ra: float = ra_field
+    dec: float = dec_field
+
+    fwhm: float = Field(ge=0)
+
+    jd: float = Field(ge=0)
+
+    fid: int = Field(ge=0)
+
+    diffimgname: str | None = Field(max_length=255, default=None)
+    sciimgname: str | None = Field(max_length=255, default=None)
+    refimgname: str | None = Field(max_length=255, default=None)
+
+    magpsf: float = Field()
+    sigmapsf: float = Field(ge=0)
+    chipsf: float | None = Field(ge=0, default=None)
+
+    aimage: float = Field(ge=0)
+    bimage: float = Field(ge=0)
+    aimagerat: float = Field(ge=0)
+    bimagerat: float = Field(ge=0)
+    elong: float = Field(ge=0)
+
+    scorr: float = Field(ge=0)
+
+    xpos: float | None = Field(ge=0, default=None)
+    ypos: float | None = Field(ge=0, default=None)
+
+    magzpsci: float | None = Field(default=None)
+    magzpsciunc: float | None = Field(ge=0, default=None)
+    magzpscirms: float | None = Field(ge=0, default=None)
+
+    tmjmag1: float | None = Field(default=None)
+    tmhmag1: float | None = Field(default=None)
+    tmkmag1: float | None = Field(default=None)
+    tmobjectid1: str | None = Field(default=None)
+
+
+if DB_USER is not None:
+    setup_database(WircBase)
diff --git a/mirar/pipelines/wirc/wirc_files/schema/candidates.sql b/mirar/pipelines/wirc/wirc_files/schema/candidates.sql
deleted file mode 100644
index 7499ef440..000000000
--- a/mirar/pipelines/wirc/wirc_files/schema/candidates.sql
+++ /dev/null
@@ -1,31 +0,0 @@
-CREATE TABLE IF NOT EXISTS candidates (
-    "candid" BIGINT PRIMARY KEY,
-    "objectId" VARCHAR(15),
-    "ra" REAL,
-    "dec" REAL,
-    "fwhm" REAL,
-    "jd" REAL,
-    "fid" INT,
-    "diffimname" VARCHAR(255),
-    "sciimname" VARCHAR(255),
-    "refimname" VARCHAR(255),
-    "magpsf" REAL,
-    "sigmapsf" REAL,
-    "chipsf" REAL,
-    "aimage" REAL,
-    "bimage" REAL,
-    "aimagerat" REAL,
-    "bimagerat" REAL,
-    "elong" REAL,
-    "psra1" REAL,
-    "psdec1" REAL,
-    "scorr" REAL,
-    "xpos" REAL,
-    "ypos" REAL,
-    "magzpsci" REAL,
-    "magzpsciunc" REAL,
-    "tmjmag1" REAL,
-    "tmhmag1" REAL,
-    "tmkmag1" REAL,
-    "tmobjectid1" VARCHAR(25)
-);
diff --git a/mirar/processors/avro/send_to_fritz.py b/mirar/processors/avro/send_to_fritz.py
index 1fb0db2f6..5f4a14054 100644
--- a/mirar/processors/avro/send_to_fritz.py
+++ b/mirar/processors/avro/send_to_fritz.py
@@ -29,7 +29,7 @@
 from urllib3.util import Retry
 
 from mirar.data import SourceBatch
-from mirar.paths import PACKAGE_NAME, __version__
+from mirar.paths import CAND_NAME_KEY, PACKAGE_NAME, __version__
 from mirar.processors.base_processor import BaseSourceProcessor
 
 matplotlib.use("agg")
@@ -277,23 +277,24 @@ def alert_post_source(self, alert: dict, group_ids: Optional[list[int]] = None):
         data = {
             "ra": alert["ra"],
             "dec": alert["dec"],
-            "id": alert["objectId"],
+            "id": alert[CAND_NAME_KEY],
             "group_ids": group_ids,
             "origin": self.origin,
         }
 
         logger.debug(
-            f"Saving {alert['objectId']} {alert['candid']} as a Source on SkyPortal"
+            f"Saving {alert[CAND_NAME_KEY]} {alert['candid']} as a Source on SkyPortal"
         )
         response = self.api("POST", "https://fritz.science/api/sources", data)
 
         if response.json()["status"] == "success":
             logger.debug(
-                f"Saved {alert['objectId']} {alert['candid']} as a Source on SkyPortal"
+                f"Saved {alert[CAND_NAME_KEY]} {alert['candid']} "
+                f"as a Source on SkyPortal"
             )
         else:
             err = (
-                f"Failed to save {alert['objectId']} {alert['candid']} "
+                f"Failed to save {alert[CAND_NAME_KEY]} {alert['candid']} "
                 f"as a Source on SkyPortal"
             )
             logger.error(err)
@@ -308,7 +309,7 @@ def alert_post_candidate(self, alert):
         """
 
         data = {
-            "id": alert["objectId"],
+            "id": alert[CAND_NAME_KEY],
             "ra": alert["ra"],
             "dec": alert["dec"],
             "filter_ids": [self.fritz_filter_id],
@@ -318,17 +319,17 @@ def alert_post_candidate(self, alert):
         }
 
         logger.debug(
-            f"Posting metadata of {alert['objectId']} {alert['candid']} to SkyPortal"
+            f"Posting metadata of {alert[CAND_NAME_KEY]} {alert['candid']} to SkyPortal"
         )
         response = self.api("POST", "https://fritz.science/api/candidates", data)
 
         if response.json()["status"] == "success":
             logger.debug(
-                f"Posted {alert['objectId']} {alert['candid']} metadata to SkyPortal"
+                f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} metadata to SkyPortal"
             )
         else:
             logger.error(
-                f"Failed to post {alert['objectId']} {alert['candid']} "
+                f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
                 f"metadata to SkyPortal"
             )
             logger.error(response.json())
@@ -388,7 +389,7 @@ def make_thumbnail(self, alert, skyportal_type: str, alert_packet_type: str):
         plt.close("all")
 
         thumbnail_dict = {
-            "obj_id": alert["objectId"],
+            "obj_id": alert[CAND_NAME_KEY],
             "data": base64.b64encode(buff.read()).decode("utf-8"),
             "ttype": skyportal_type,
         }
@@ -407,25 +408,25 @@ def alert_post_thumbnails(self, alert):
             ("sub", "Difference"),
         ]:
             logger.debug(
-                f"Making {instrument_type} thumbnail for {alert['objectId']} "
+                f"Making {instrument_type} thumbnail for {alert[CAND_NAME_KEY]} "
                 f"{alert['candid']}",
             )
             thumb = self.make_thumbnail(alert, ttype, instrument_type)
 
             logger.debug(
-                f"Posting {instrument_type} thumbnail for {alert['objectId']} "
+                f"Posting {instrument_type} thumbnail for {alert[CAND_NAME_KEY]} "
                 f"{alert['candid']} to SkyPortal",
             )
             response = self.api("POST", "https://fritz.science/api/thumbnail", thumb)
 
             if response.json()["status"] == "success":
                 logger.debug(
-                    f"Posted {alert['objectId']} {alert['candid']} "
+                    f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} "
                     f"{instrument_type} cutout to SkyPortal"
                 )
             else:
                 logger.error(
-                    f"Failed to post {alert['objectId']} {alert['candid']} "
+                    f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
                     f"{instrument_type} cutout to SkyPortal"
                 )
                 logger.error(response.json())
@@ -463,7 +464,7 @@ def upload_thumbnail(self, alert):
             cutoutb64_string = cutoutb64.decode("utf8")
 
             data_payload = {
-                "obj_id": alert["objectId"],
+                "obj_id": alert[CAND_NAME_KEY],
                 "data": cutoutb64_string,
                 "ttype": fritz_key,
             }
@@ -474,19 +475,19 @@ def upload_thumbnail(self, alert):
 
             if response.json()["status"] == "success":
                 logger.debug(
-                    f"Posted {alert['objectId']} {alert['candid']} "
+                    f"Posted {alert[CAND_NAME_KEY]} {alert['candid']} "
                     f"{cand_key} cutout to SkyPortal"
                 )
             else:
                 logger.error(
-                    f"Failed to post {alert['objectId']} {alert['candid']} "
+                    f"Failed to post {alert[CAND_NAME_KEY]} {alert['candid']} "
                     f"{cand_key} cutout to SkyPortal"
                 )
                 logger.error(response.json())
 
     def make_photometry(self, alert, jd_start: Optional[float] = None):
         """
-        Make a de-duplicated pandas.DataFrame with photometry of alert['objectId']
+        Make a de-duplicated pandas.DataFrame with photometry of alert[CAND_NAME_KEY]
         Modified from Kowalksi (https://github.com/dmitryduev/kowalski)
 
         :param alert: candidate dictionary
@@ -496,7 +497,7 @@ def make_photometry(self, alert, jd_start: Optional[float] = None):
         top_level = [
             "schemavsn",
             "publisher",
-            "objectId",
+            CAND_NAME_KEY,
             "candid",
             "candidate",
             "prv_candidates",
@@ -592,13 +593,13 @@ def make_photometry(self, alert, jd_start: Optional[float] = None):
     def alert_put_photometry(self, alert):
         """Send photometry to Fritz."""
         logger.debug(
-            f"Making alert photometry of {alert['objectId']} {alert['candid']}"
+            f"Making alert photometry of {alert[CAND_NAME_KEY]} {alert['candid']}"
         )
         df_photometry = self.make_photometry(alert)
 
         # post photometry
         photometry = {
-            "obj_id": alert["objectId"],
+            "obj_id": alert[CAND_NAME_KEY],
             "stream_ids": [int(self.stream_id)],
             "instrument_id": self.instrument_id,
             "mjd": df_photometry["mjd"].tolist(),
@@ -615,7 +616,7 @@ def alert_put_photometry(self, alert):
             len(photometry.get("fluxerr", ())) > 0
         ):
             logger.debug(
-                f"Posting photometry of {alert['objectId']} {alert['candid']}, "
+                f"Posting photometry of {alert[CAND_NAME_KEY]} {alert['candid']}, "
                 f"stream_id={self.stream_id} to SkyPortal"
             )
             response = self.api(
@@ -623,12 +624,12 @@ def alert_put_photometry(self, alert):
             )
             if response.json()["status"] == "success":
                 logger.debug(
-                    f"Posted {alert['objectId']} photometry stream_id={self.stream_id} "
+                    f"Posted {alert[CAND_NAME_KEY]} photometry stream_id={self.stream_id} "
                     f"to SkyPortal"
                 )
             else:
                 logger.error(
-                    f"Failed to post {alert['objectId']} photometry "
+                    f"Failed to post {alert[CAND_NAME_KEY]} photometry "
                     f"stream_id={self.stream_id} to SkyPortal"
                 )
                 logger.error(response.json())
@@ -642,27 +643,32 @@ def alert_post_annotation(self, alert):
         }
         payload = {"origin": self.origin, "data": data, "group_ids": self.group_ids}
 
-        path = f'https://fritz.science/api/sources/{str(alert["objectId"])}/annotations'
+        path = (
+            f"https://fritz.science/api/sources/{str(alert[CAND_NAME_KEY])}/annotations"
+        )
         response = self.api("POST", path, payload)
 
         if response.json()["status"] == "success":
-            logger.debug(f"Posted {alert['objectId']} annotation to SkyPortal")
+            logger.debug(f"Posted {alert[CAND_NAME_KEY]} annotation to SkyPortal")
         else:
-            logger.error(f"Failed to post {alert['objectId']} annotation to SkyPortal")
+            logger.error(
+                f"Failed to post {alert[CAND_NAME_KEY]} annotation to SkyPortal"
+            )
             logger.error(response.json())
 
     def alert_put_annotation(self, alert):
         """Retrieve an annotation to check if it exists already."""
         response = self.api(
             "GET",
-            f'https://fritz.science/api/sources/{str(alert["objectId"])}/annotations',
+            f"https://fritz.science/api/sources/"
+            f"{str(alert[CAND_NAME_KEY])}/annotations",
         )
 
         if response.json()["status"] == "success":
-            logger.debug(f"Got {alert['objectId']} annotations from SkyPortal")
+            logger.debug(f"Got {alert[CAND_NAME_KEY]} annotations from SkyPortal")
         else:
             logger.debug(
-                f"Failed to get {alert['objectId']} annotations from SkyPortal"
+                f"Failed to get {alert[CAND_NAME_KEY]} annotations from SkyPortal"
             )
             logger.debug(response.json())
             return False
@@ -688,29 +694,29 @@ def alert_put_annotation(self, alert):
             }
             new_annotation = {
                 "author_id": existing_annotations[self.origin]["author_id"],
-                "obj_id": alert["objectId"],
+                "obj_id": alert[CAND_NAME_KEY],
                 "origin": self.origin,
                 "data": data,
                 "group_ids": self.group_ids,
             }
 
             logger.debug(
-                f"Putting annotation for {alert['objectId']} {alert['candid']} "
+                f"Putting annotation for {alert[CAND_NAME_KEY]} {alert['candid']} "
                 f"to SkyPortal",
             )
             response = self.api(
                 "PUT",
-                f"https://fritz.science/api/sources/{alert['objectId']}"
+                f"https://fritz.science/api/sources/{alert[CAND_NAME_KEY]}"
                 f"/annotations/{existing_annotations[self.origin]['annotation_id']}",
                 new_annotation,
             )
             if response.json()["status"] == "success":
                 logger.debug(
-                    f"Posted updated {alert['objectId']} annotation to SkyPortal"
+                    f"Posted updated {alert[CAND_NAME_KEY]} annotation to SkyPortal"
                 )
             else:
                 logger.error(
-                    f"Failed to post updated {alert['objectId']} annotation "
+                    f"Failed to post updated {alert[CAND_NAME_KEY]} annotation "
                     f"to SkyPortal"
                 )
                 logger.error(response.json())
@@ -724,24 +730,25 @@ def alert_skyportal_manager(self, alert):
         :type alert: _type_
         """
         # check if candidate exists in SkyPortal
-        logger.debug(f"Checking if {alert['objectId']} is candidate in SkyPortal")
+        logger.debug(f"Checking if {alert[CAND_NAME_KEY]} is candidate in SkyPortal")
         response = self.api(
-            "HEAD", f"https://fritz.science/api/candidates/{alert['objectId']}"
+            "HEAD", f"https://fritz.science/api/candidates/{alert[CAND_NAME_KEY]}"
         )
         is_candidate = response.status_code == 200
         logger.debug(
-            f"{alert['objectId']} {'is' if is_candidate else 'is not'} "
+            f"{alert[CAND_NAME_KEY]} {'is' if is_candidate else 'is not'} "
             f"candidate in SkyPortal"
         )
 
         # check if source exists in SkyPortal
-        logger.debug(f"Checking if {alert['objectId']} is source in SkyPortal")
+        logger.debug(f"Checking if {alert[CAND_NAME_KEY]} is source in SkyPortal")
         response = self.api(
-            "HEAD", f"https://fritz.science/api/sources/{alert['objectId']}"
+            "HEAD", f"https://fritz.science/api/sources/{alert[CAND_NAME_KEY]}"
         )
         is_source = response.status_code == 200
         logger.debug(
-            f"{alert['objectId']} {'is' if is_source else 'is not'} source in SkyPortal"
+            f"{alert[CAND_NAME_KEY]} "
+            f"{'is' if is_source else 'is not'} source in SkyPortal"
         )
 
         # object does not exist in SkyPortal: neither cand nor source
@@ -774,11 +781,12 @@ def alert_skyportal_manager(self, alert):
             if is_source:
                 # get info on the corresponding groups:
                 logger.debug(
-                    f"Getting source groups info on {alert['objectId']} from SkyPortal",
+                    f"Getting source groups info on "
+                    f"{alert[CAND_NAME_KEY]} from SkyPortal",
                 )
                 response = self.api(
                     "GET",
-                    f"https://fritz.science/api/sources/{alert['objectId']}/groups",
+                    f"https://fritz.science/api/sources/{alert[CAND_NAME_KEY]}/groups",
                 )
                 if response.json()["status"] == "success":
                     existing_groups = response.json()["data"]
@@ -789,7 +797,7 @@ def alert_skyportal_manager(self, alert):
                             self.alert_post_source(alert, [existing_gid])
                 else:
                     logger.error(
-                        f"Failed to get source groups info on {alert['objectId']}"
+                        f"Failed to get source groups info on {alert[CAND_NAME_KEY]}"
                     )
             else:  # exists in SkyPortal but NOT saved as a source
                 self.alert_post_source(alert)
@@ -800,7 +808,7 @@ def alert_skyportal_manager(self, alert):
             if self.update_thumbnails:
                 self.alert_post_thumbnails(alert)
 
-        logger.debug(f'SendToFritz Manager complete for {alert["objectId"]}')
+        logger.debug(f"SendToFritz Manager complete for {alert[CAND_NAME_KEY]}")
 
     def make_alert(self, candidate_df: pd.DataFrame):
         """
diff --git a/mirar/processors/database/__init__.py b/mirar/processors/database/__init__.py
index 165858de2..1df8ca647 100644
--- a/mirar/processors/database/__init__.py
+++ b/mirar/processors/database/__init__.py
@@ -1,9 +1,13 @@
 """
 Central module for interactions with postgres databases
 """
-from mirar.processors.database.base_database_processor import BaseDatabaseProcessor
-from mirar.processors.database.postgres import (
-    POSTGRES_DUPLICATE_PROTOCOLS,
-    PostgresAdmin,
-    PostgresUser,
+from mirar.processors.database.database_inserter import (
+    DatabaseImageInserter,
+    DatabaseSourceInserter,
 )
+from mirar.processors.database.database_selector import (
+    DatabaseCrossmatchSelector,
+    DatabaseHistorySelector,
+    DatabaseSourceSelector,
+)
+from mirar.processors.database.database_updater import ImageDatabaseUpdater
diff --git a/mirar/processors/database/base_database_processor.py b/mirar/processors/database/base_database_processor.py
index 9187752be..86de68de6 100644
--- a/mirar/processors/database/base_database_processor.py
+++ b/mirar/processors/database/base_database_processor.py
@@ -3,17 +3,12 @@
 """
 import logging
 from abc import ABC
-from pathlib import Path
-from typing import Optional
+from typing import Type
 
-from mirar.paths import max_n_cpu
+from mirar.database.base_model import BaseDB
+from mirar.database.constants import POSTGRES_DUPLICATE_PROTOCOLS
+from mirar.database.user import PostgresAdmin, PostgresUser
 from mirar.processors.base_processor import BaseProcessor
-from mirar.processors.database.postgres import (
-    POSTGRES_DUPLICATE_PROTOCOLS,
-    DataBaseError,
-    PostgresAdmin,
-    PostgresUser,
-)
 
 logger = logging.getLogger(__name__)
 
@@ -21,170 +16,21 @@
 class BaseDatabaseProcessor(BaseProcessor, ABC):
     """Base class for processors which interact with a postgres database"""
 
-    max_n_cpu = min(max_n_cpu, 5)
+    max_n_cpu = 1
 
     def __init__(
         self,
-        db_name: str,
-        db_table: str,
-        schema_path: str | Path,
+        db_table: Type[BaseDB],
         pg_user: PostgresUser = PostgresUser(),
         pg_admin: PostgresAdmin = PostgresAdmin(),
-        has_foreign_keys: bool = False,
-        schema_dir: Optional[str | Path] = None,
         duplicate_protocol: str = "fail",
-        q3c_bool: bool = False,
     ):
         super().__init__()
-        self.db_name = db_name
         self.db_table = db_table
-
-        self.schema_path = schema_path
-        self.schema_dir = Path(schema_dir) if schema_dir is not None else None
+        self.db_name = self.db_table.sql_model.db_name
+        self.duplicate_protocol = duplicate_protocol
 
         self.pg_user = pg_user
         self._pg_admin = pg_admin
 
-        self.has_foreign_keys = has_foreign_keys
-        self.db_check_bool = False
-        self.duplicate_protocol = duplicate_protocol
-        self.q3c = q3c_bool
-
         assert self.duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS
-
-    def db_exists(self) -> bool:
-        """
-        Checks if a database exists
-
-        :return: boolean
-        """
-        return self._pg_admin.check_if_db_exists(db_name=self.db_name)
-
-    def _make_db(self):
-        """
-        Creates a database
-
-        :return:
-        """
-        self._pg_admin.create_db(db_name=self.db_name)
-
-    def pg_user_exists(self):
-        """
-        Check if the specified db user exists
-
-        :return: boolean
-        """
-        return self._pg_admin.check_if_user_exists(self.pg_user.db_user)
-
-    def _make_pg_user(self):
-        """
-        Creates a new database user
-
-        :return: None
-        """
-        self._pg_admin.create_new_user(
-            new_db_user=self.pg_user.db_user, new_password=self.pg_user.db_password
-        )
-
-    def _grant_privileges_to_pg_user(self):
-        """
-        Grants db privilege to user
-
-        :return: None
-        """
-        self._pg_admin.grant_privileges(self.db_name, db_user=self.pg_user.db_user)
-
-    def table_exists(self):
-        """
-        Check if the database table exists
-
-        :return: boolean
-        """
-        return self.pg_user.check_if_table_exists(
-            db_name=self.db_name,
-            db_table=self.db_table,
-        )
-
-    def _create_table(self, schema_path: str | Path):
-        """
-        Makes a database table using schema
-
-        :param schema_path: Path of schema file
-        :return: None
-        """
-        self.pg_user.create_table(
-            schema_path,
-            db_name=self.db_name,
-        )
-
-    def check_prerequisites(
-        self,
-    ):
-        if not self.db_check_bool:
-            self.check_database_setup()
-            self.db_check_bool = True
-
-    def check_database_setup(self):
-        """
-        Checks the database is set up, and if not, creates the database and tables
-
-        :return: None
-        """
-        # If not specified, admin credentials equal user credentials
-        # Check that the credentials work at all
-
-        self._pg_admin.validate_credentials()
-
-        # Create pg user for the first time
-
-        if not self.pg_user_exists():
-            self._make_pg_user()
-        self.pg_user.validate_credentials()
-
-        # Check there is a db
-
-        if not self.db_exists():
-            self._make_db()
-            self._grant_privileges_to_pg_user()
-
-            if self.has_foreign_keys:
-                if self.schema_dir is None:
-                    self.schema_dir = self.schema_path.parent
-                    logger.warning(
-                        f"Warning, integrated db setup with foreign keys requested, "
-                        f"but no schema directory specified. "
-                        f"Will search for schema files in {self.schema_dir} directory."
-                    )
-
-                logger.debug(
-                    f"Looking in {self.schema_dir} directory to search for schema files"
-                )
-
-                self.pg_user.create_tables_from_schema(self.schema_dir, self.db_name)
-
-                if self.q3c:
-                    q3c_dir = self.schema_dir.joinpath("q3c")
-                    q3c_indexes_file = q3c_dir.joinpath("q3c_indexes.sql")
-                    self.pg_user.run_sql_command_from_file(
-                        file_path=q3c_indexes_file, db_name=self.db_name
-                    )
-                    logger.debug("Created q3c_bool indexes")
-
-        if not self.table_exists():
-            self._create_table(self.schema_path)
-
-            if self.q3c:
-                q3c_dir = self.schema_dir.joinpath("q3c")
-                table_q3c_path = q3c_dir.joinpath(f"q3c_{self.db_table}.sql")
-
-                if not table_q3c_path.exists():
-                    err = (
-                        f"q3c_bool extension requested but no "
-                        f"{table_q3c_path} file found. Please add it in."
-                    )
-                    logger.error(err)
-                    raise DataBaseError(err)
-
-                self.pg_user.run_sql_command_from_file(
-                    file_path=table_q3c_path, db_name=self.db_name
-                )
diff --git a/mirar/processors/database/constraints.py b/mirar/processors/database/constraints.py
deleted file mode 100644
index 00b8576d9..000000000
--- a/mirar/processors/database/constraints.py
+++ /dev/null
@@ -1,106 +0,0 @@
-"""
-Module for DBQueryConstraints to carefully specify postgres query constraints
-"""
-from typing import Optional
-
-import numpy as np
-
-POSTGRES_ACCEPTED_COMPARISONS = ["=", "<", ">", "between"]
-
-
-class DBQueryConstraints:
-    """
-    Object containing one or more postgres query constraints
-    """
-
-    def __init__(
-        self,
-        columns: str | list[str],
-        accepted_values: str | int | float | list[str | float | int | list],
-        comparison_types: Optional[str | list[str]] = None,
-    ):
-        self.columns = []
-        self.accepted_values = []
-        self.comparison_types = []
-
-        if not isinstance(columns, list):
-            columns = [columns]
-        if not isinstance(accepted_values, list):
-            accepted_values = [accepted_values]
-
-        assert len(columns) == len(accepted_values)
-
-        if comparison_types is None:
-            comparison_types = ["="] * len(accepted_values)
-        assert len(comparison_types) == len(accepted_values)
-
-        for i, column in enumerate(columns):
-            self.add_constraint(
-                column=column,
-                accepted_values=accepted_values[i],
-                comparison_type=comparison_types[i],
-            )
-
-    def add_constraint(
-        self,
-        column: str,
-        accepted_values: str | int | float | tuple[float, float] | tuple[int, int],
-        comparison_type: str = "=",
-    ):
-        """
-        Add a new constraint
-
-        :param column: column
-        :param accepted_values: accepted value for comparison
-        :param comparison_type: type of comparison, e.g '='
-        :return: None
-        """
-        assert comparison_type in POSTGRES_ACCEPTED_COMPARISONS
-
-        if comparison_type == "between":
-            assert np.logical_and(
-                isinstance(accepted_values, tuple), len(accepted_values) == 2
-            )
-
-        self.columns.append(column)
-        self.accepted_values.append(accepted_values)
-        self.comparison_types.append(comparison_type)
-
-    def __add__(self, other):
-        new = self.__class__(self.columns, self.accepted_values, self.comparison_types)
-        for args in other:
-            new.add_constraint(*args)
-        return new
-
-    def __iadd__(self, other):
-        for args in other:
-            self.add_constraint(*args)
-        return self
-
-    def __len__(self):
-        return self.columns.__len__()
-
-    def __iter__(self):
-        return iter(zip(self.columns, self.accepted_values, self.comparison_types))
-
-    def parse_constraints(
-        self,
-    ) -> str:
-        """
-        Converts the list of constraints to sql
-
-        :return: sql string
-        """
-        constraints = []
-        for i, column in enumerate(self.columns):
-            if self.comparison_types[i] == "between":
-                constraints.append(
-                    f"{column} between {self.accepted_values[i][0]} "
-                    f"and {self.accepted_values[i][1]}"
-                )
-            else:
-                constraints.append(
-                    f"{column} {self.comparison_types[i]} {self.accepted_values[i]}"
-                )
-
-        return " AND ".join(constraints)
diff --git a/mirar/processors/database/database_exporter.py b/mirar/processors/database/database_exporter.py
deleted file mode 100644
index 4896973a5..000000000
--- a/mirar/processors/database/database_exporter.py
+++ /dev/null
@@ -1,76 +0,0 @@
-"""
-Module containing processors for exporting to databases
-"""
-import logging
-from abc import ABC
-
-from mirar.data import ImageBatch, SourceBatch
-from mirar.processors.base_processor import BaseImageProcessor, BaseSourceProcessor
-from mirar.processors.database.base_database_processor import BaseDatabaseProcessor
-
-logger = logging.getLogger(__name__)
-
-
-class BaseDatabaseExporter(BaseDatabaseProcessor, ABC):
-    """
-    Base class for DB exporters
-    """
-
-    base_key = "dbexporter"
-
-    def __str__(self):
-        return (
-            f"Processor to save "
-            f"{['candidates', 'images'][isinstance(self, BaseImageProcessor)]} "
-            f"to the '{self.db_table}' table of the '{self.db_name}' Postgres database."
-        )
-
-
-class DatabaseImageExporter(BaseDatabaseExporter, BaseImageProcessor):
-    """
-    Processor for exporting images to a database
-    """
-
-    def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:
-        for image in batch:
-            primary_keys, primary_key_values = self.pg_user.export_to_db(
-                image,
-                db_name=self.db_name,
-                db_table=self.db_table,
-                duplicate_protocol=self.duplicate_protocol,
-            )
-
-            for ind, key in enumerate(primary_keys):
-                image[key] = primary_key_values[ind]
-        return batch
-
-
-class DatabaseDataframeExporter(BaseDatabaseExporter, BaseSourceProcessor):
-    """
-    Processor for exporting sources to a database
-    """
-
-    def _apply_to_sources(self, batch: SourceBatch) -> SourceBatch:
-        for source_list in batch:
-            candidate_table = source_list.get_data()
-
-            primary_key_dict = {}
-            for _, candidate_row in candidate_table.iterrows():
-                primary_keys, primary_key_values = self.pg_user.export_to_db(
-                    candidate_row.to_dict(),
-                    db_name=self.db_name,
-                    db_table=self.db_table,
-                    duplicate_protocol=self.duplicate_protocol,
-                )
-                for ind, key in enumerate(primary_keys):
-                    if key not in primary_key_dict:
-                        primary_key_dict[key] = [primary_key_values[ind]]
-                    else:
-                        primary_key_dict[key].append(primary_key_values[ind])
-
-            for key, val in primary_key_dict.items():
-                candidate_table[key] = val
-
-            source_list.set_data(candidate_table)
-
-        return batch
diff --git a/mirar/processors/sqldatabase/database_exporter.py b/mirar/processors/database/database_inserter.py
similarity index 55%
rename from mirar/processors/sqldatabase/database_exporter.py
rename to mirar/processors/database/database_inserter.py
index a21b52663..625c4178e 100644
--- a/mirar/processors/sqldatabase/database_exporter.py
+++ b/mirar/processors/database/database_inserter.py
@@ -5,11 +5,12 @@
 from abc import ABC
 
 import numpy as np
+import pandas as pd
 
-from mirar.data import ImageBatch, SourceBatch
+from mirar.data import Image, ImageBatch, SourceBatch
 from mirar.errors.exceptions import BaseProcessorError
 from mirar.processors.base_processor import BaseImageProcessor, BaseSourceProcessor
-from mirar.processors.sqldatabase.base_database_processor import BaseDatabaseProcessor
+from mirar.processors.database.base_database_processor import BaseDatabaseProcessor
 from mirar.processors.utils.image_selector import ImageBatcher
 
 logger = logging.getLogger(__name__)
@@ -21,12 +22,12 @@ class ImageBatchDatabaseExporterError(BaseProcessorError):
     """
 
 
-class BaseDatabaseExporter(BaseDatabaseProcessor, ABC):
+class BaseDatabaseInserter(BaseDatabaseProcessor, ABC):
     """
-    Base class for DB exporters
+    Base class for DB inserter
     """
 
-    base_key = "dbexporter"
+    base_key = "dbinserter"
     max_n_cpu = 1
 
     def __str__(self):
@@ -38,57 +39,78 @@ def __str__(self):
         )
 
 
-class DatabaseImageExporter(BaseDatabaseExporter, BaseImageProcessor):
+class DatabaseImageInserter(BaseDatabaseInserter, BaseImageProcessor):
     """
     Processor for exporting images to a database
     """
 
     def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:
         for image in batch:
-            primary_keys, primary_key_values = self.pg_user.export_to_db(
-                image,
-                db_table=self.db_table,
-                duplicate_protocol=self.duplicate_protocol,
-            )
+            val_dict = self.generate_value_dict(image)
+
+            new = self.db_table(**val_dict)
+            res = new.insert_entry()
 
-            for ind, key in enumerate(primary_keys):
-                image[key] = primary_key_values[ind]
+            assert len(res) == 1
+
+            for key in res.columns:
+                image[key] = res[key].iloc[0]
         return batch
 
+    @staticmethod
+    def generate_value_dict(image: Image) -> dict:
+        """
+        Generate a dictionary of lower case keys and values from an image
+
+        :param image: Image
+        :return: Dictionary of lower case keys and values
+        """
+        return {key.lower(): image[key] for key in image.keys()}
 
-class DatabaseDataframeExporter(BaseDatabaseExporter, BaseSourceProcessor):
+
+class DatabaseSourceInserter(BaseDatabaseInserter, BaseSourceProcessor):
     """
     Processor for exporting sources to a database
     """
 
     def _apply_to_sources(self, batch: SourceBatch) -> SourceBatch:
         for source_list in batch:
-            candidate_table = source_list.get_data()
+            source_table = source_list.get_data()
+            metadata = source_list.get_metadata()
 
             primary_key_dict = {}
-            for _, candidate_row in candidate_table.iterrows():
-                primary_keys, primary_key_values = self.pg_user.export_to_db(
-                    candidate_row.to_dict(),
-                    db_table=self.db_table,
-                )
-                for ind, key in enumerate(primary_keys):
-                    if key not in primary_key_dict:
-                        primary_key_dict[key] = [primary_key_values[ind]]
-                    else:
-                        primary_key_dict[key].append(primary_key_values[ind])
-                    # candidate_row[key] = primary_key_values[ind]
-
-                # new_table = pd.concat([new_table,candidate_row])
-                # new_table.append(candidate_row, ignore_index=True)
+            for _, source_row in source_table.iterrows():
+                super_dict = self.generate_super_dict(metadata, source_row)
+
+                new = self.db_table(**super_dict)
+                res = new.insert_entry()
+
+                assert len(res) == 1
+
             for key, val in primary_key_dict.items():
-                candidate_table[key] = val
+                source_table[key] = val
 
-            source_list.set_data(candidate_table)
+            source_list.set_data(source_table)
 
         return batch
 
+    @staticmethod
+    def generate_super_dict(metadata: dict, source_row: pd.Series) -> dict:
+        """
+        Generate a dictionary of metadata and candidate row, with lower case keys
+
+        :param metadata: Metadata for the source table
+        :param source_row: Individual row of the source table
+        :return: Combined dictionary
+        """
+        super_dict = {key.lower(): val for key, val in metadata.items()}
+        super_dict.update(
+            {key.lower(): val for key, val in source_row.to_dict().items()}
+        )
+        return super_dict
+
 
-class DatabaseImageBatchExporter(DatabaseImageExporter):
+class DatabaseImageBatchInserter(DatabaseImageInserter):
     """
     Processor for creating a single entry per batch of images in a database
     """
@@ -119,15 +141,19 @@ def _apply_to_images(self, batch: ImageBatch) -> ImageBatch:
 
         image = batch[0]
         logger.debug(f"Trying to export {[image[x] for x in column_names]}")
-        primary_keys, primary_key_values = self.pg_user.export_to_db(
-            image,
-            db_table=self.db_table,
-            duplicate_protocol=self.duplicate_protocol,
-        )
 
-        for ind, key in enumerate(primary_keys):
+        val_dict = {key.lower(): image[key] for key in image.keys()}
+
+        new = self.db_table(**val_dict)
+        res = new.insert_entry()
+
+        assert len(res) == 1
+
+        for key in res.columns:
+            val = res[key].iloc[0]
             for image in batch:
-                image[key] = primary_key_values[ind]
+                image[key] = val
+
         return batch
 
     def check_prerequisites(
diff --git a/mirar/processors/database/database_modifier.py b/mirar/processors/database/database_modifier.py
deleted file mode 100644
index 02ed009b0..000000000
--- a/mirar/processors/database/database_modifier.py
+++ /dev/null
@@ -1,128 +0,0 @@
-"""
-Module for processors to modify database entries
-"""
-import logging
-from abc import ABC
-from typing import Optional
-
-import numpy as np
-import pandas as pd
-
-from mirar.data import ImageBatch
-from mirar.processors.database.constraints import DBQueryConstraints
-from mirar.processors.database.database_importer import (
-    BaseDatabaseImporter,
-    BaseImageDatabaseImporter,
-)
-
-logger = logging.getLogger(__name__)
-
-
-class BaseDatabaseModifier(BaseDatabaseImporter, ABC):
-    """
-    Base class for database modifiers
-    """
-
-    base_key = "dbmodifier"
-
-    def __init__(self, db_alter_columns: Optional[str] = None, **kwargs):
-        super().__init__(db_output_columns=db_alter_columns, **kwargs)
-        self.db_alter_columns = db_alter_columns
-
-
-class ImageDatabaseModifier(BaseDatabaseModifier, BaseImageDatabaseImporter, ABC):
-    """Base Class for modifying image entries in a database"""
-
-    def _apply_to_images(
-        self,
-        batch: ImageBatch,
-    ) -> ImageBatch:
-        for image in batch:
-            query_constraints = self.get_constraints(image)
-
-            self.pg_user.modify_db_entry(
-                value_dict=image,
-                db_constraints=query_constraints,
-                db_alter_columns=self.db_alter_columns,
-                db_table=self.db_table,
-                db_name=self.db_name,
-            )
-
-        return batch
-
-
-class ModifyImageDatabaseSeq(ImageDatabaseModifier):
-    """Processor to modify images in a database with a sequence"""
-
-    def __init__(self, sequence_key: Optional[str | list[str]] = None, **kwargs):
-        super().__init__(**kwargs)
-        self.sequence_key = sequence_key
-
-    def get_constraints(self, data):
-        if self.sequence_key is None:
-            self.sequence_key = list(
-                self.pg_user.get_sequence_keys_from_table(self.db_table, self.db_name)
-            )
-
-        accepted_values = [data[x.upper()] for x in self.sequence_key]
-        comparison_types = ["="] * len(accepted_values)
-
-        query_constraints = DBQueryConstraints(
-            columns=self.sequence_key,
-            accepted_values=accepted_values,
-            comparison_types=comparison_types,
-        )
-
-        return query_constraints
-
-
-class ModifyImageDatabaseSeqList(ModifyImageDatabaseSeq):
-    """
-    Processor to modify multiple entries specified by a list of sequences in an
-    image database
-    """
-
-    def __init__(self, sequence_key: Optional[str | list[str]], **kwargs):
-        super().__init__(**kwargs)
-        self.sequence_key = sequence_key
-        if isinstance(self.sequence_key, str):
-            self.sequence_key = [self.sequence_key]
-        if self.sequence_key is None:
-            self.sequence_key = list(
-                self.pg_user.get_sequence_keys_from_table(self.db_table, self.db_name)
-            )
-
-    def _apply_to_images(
-        self,
-        batch: ImageBatch,
-    ) -> ImageBatch:
-        for image in batch:
-            try:
-                data_array = [
-                    [int(y) for y in image[x.upper()].split(",")]
-                    for x in self.sequence_key
-                ]
-            except ValueError as exc:
-                raise ValueError("Sequence keys must be integers") from exc
-
-            data_array_lens = [len(x) for x in data_array]
-            if len(set(data_array_lens)) > 1:
-                raise ValueError("All sequence keys must have the same length")
-
-            data_df = pd.DataFrame(
-                np.array(data_array).T, columns=[x.upper() for x in self.sequence_key]
-            )
-            logger.debug(data_df)
-            for ind in range(len(data_df)):
-                row = data_df.iloc[ind]
-                query_constraints = self.get_constraints(row)
-
-                self.pg_user.modify_db_entry(
-                    value_dict=image,
-                    db_constraints=query_constraints,
-                    db_alter_columns=self.db_alter_columns,
-                    db_table=self.db_table,
-                    db_name=self.db_name,
-                )
-
-        return batch
diff --git a/mirar/processors/database/database_importer.py b/mirar/processors/database/database_selector.py
similarity index 54%
rename from mirar/processors/database/database_importer.py
rename to mirar/processors/database/database_selector.py
index 1d3eb578b..b99df6ae1 100644
--- a/mirar/processors/database/database_importer.py
+++ b/mirar/processors/database/database_selector.py
@@ -9,19 +9,26 @@
 import pandas as pd
 
 from mirar.data import DataBlock, Image, ImageBatch, SourceBatch
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.transactions import select_from_table
+from mirar.paths import SOURCE_HISTORY_KEY
 from mirar.processors.base_processor import BaseImageProcessor, BaseSourceProcessor
 from mirar.processors.database.base_database_processor import BaseDatabaseProcessor
-from mirar.processors.database.constraints import DBQueryConstraints
 
 logger = logging.getLogger(__name__)
 
 
-class BaseDatabaseImporter(BaseDatabaseProcessor, ABC):
-    """Base Class for any database importer"""
+class BaseDatabaseSelector(BaseDatabaseProcessor, ABC):
+    """Base Class for any database selector"""
 
-    base_key = "dbimporter"
+    base_key = "dbselector"
 
-    def __init__(self, *args, boolean_match_key: Optional[str] = None, **kwargs):
+    def __init__(
+        self,
+        *args,
+        boolean_match_key: Optional[str] = None,
+        **kwargs,
+    ):
         super().__init__(*args, **kwargs)
         self.boolean_match_key = boolean_match_key
 
@@ -42,7 +49,7 @@ def update_header_with_single_match(data: DataBlock, res: list[dict]) -> DataBlo
     return data
 
 
-class BaseImageDatabaseImporter(BaseDatabaseImporter, BaseImageProcessor):
+class BaseImageDatabaseSelector(BaseDatabaseSelector, BaseImageProcessor):
     """
     Processor to import data from images
     """
@@ -68,12 +75,10 @@ def _apply_to_images(
         for i, image in enumerate(batch):
             query_constraints = self.get_constraints(image)
 
-            res = self.pg_user.import_from_db(
-                db_name=self.db_name,
-                db_table=self.db_table,
+            res = select_from_table(
+                sql_table=self.db_table,
                 db_constraints=query_constraints,
-                db_output_columns=self.db_output_columns,
-                output_alias_map=self.output_alias_map,
+                output_columns=self.db_output_columns,
             )
 
             image = self.update_header(image, res)
@@ -95,7 +100,7 @@ def get_constraints(self, data: DataBlock) -> None | DBQueryConstraints:
         raise NotImplementedError()
 
 
-class CrossmatchDatabaseWithHeader(BaseImageDatabaseImporter):
+class CrossmatchDatabaseWithHeader(BaseImageDatabaseSelector):
     """Processor to crossmatch to a database"""
 
     def __init__(self, db_query_columns: str | list[str], *args, **kwargs):
@@ -130,20 +135,7 @@ def get_constraints(self, data: DataBlock) -> DBQueryConstraints:
         return query_constraints
 
 
-# def update_dataframe_with_single_match(
-#     candidate_table: pd.DataFrame, results: list[dict]
-# ) -> pd.DataFrame:
-#     for res in results:
-#         assert len(res) < 1
-#
-#     keys = results[0].keys()
-#     for key in keys:
-#         candidate_table[key] = [x[0][key] for x in results]
-#
-#     return candidate_table
-
-
-class DatabaseDataframeImporter(BaseDatabaseImporter, BaseSourceProcessor, ABC):
+class DatabaseSourceSelector(BaseDatabaseSelector, BaseSourceProcessor, ABC):
     """
     Base Class for dataframe DB importers
     """
@@ -151,52 +143,15 @@ class DatabaseDataframeImporter(BaseDatabaseImporter, BaseSourceProcessor, ABC):
     def __init__(
         self,
         db_output_columns: str | list[str],
-        output_alias_map: Optional[str | list[str]] = None,
         max_num_results: Optional[int] = None,
         **kwargs,
     ):
         self.db_output_columns = db_output_columns
-        self.output_alias_map = output_alias_map
         self.max_num_results = max_num_results
         super().__init__(**kwargs)
 
-    # def _apply_to_candidates(
-    #     self,
-    #     batch: SourceBatch,
-    # ) -> SourceBatch:
-    #
-    #     for source_table in batch:
-    #         candidate_table = source_table.get_data()
-    #         results = []
-    #         for ind in range(len(candidate_table)):
-    #             cand = candidate_table.loc[ind]
-    #             (
-    #                 query_columns,
-    #                 comparison_values,
-    #                 comparison_types,
-    #             ) = self.get_constraints(cand)
-    #             res = import_from_db(
-    #                 db_name=self.db_name,
-    #                 db_table=self.db_table,
-    #                 columns=query_columns,
-    #                 accepted_values=comparison_values,
-    #                 comparison_types=comparison_types,
-    #                 db_output_columns=self.db_output_columns,
-    #                 output_alias_map=self.output_alias_map,
-    #                 db_user=self.db_user,
-    #                 password=self.db_password,
-    #                 max_num_results=self.max_num_results,
-    #             )
-    #             results.append(res)
-    #         new_df = self.update_dataframe(candidate_table, results)
-    #         source_table.set_data(new_df)
-    #     return batch
-    #
-    # def get_constraints(self, batch: SourceBatch):
-    #     raise NotImplementedError
-
-
-class DatabaseCrossmatchImporter(DatabaseDataframeImporter, BaseSourceProcessor):
+
+class DatabaseCrossmatchSelector(DatabaseSourceSelector, BaseSourceProcessor):
     """
     Processor to crossmatch to sources in a database
     """
@@ -204,14 +159,10 @@ class DatabaseCrossmatchImporter(DatabaseDataframeImporter, BaseSourceProcessor)
     def __init__(
         self,
         crossmatch_radius_arcsec: float,
-        user_defined_constraints: Optional[
-            Callable[[pd.DataFrame], DBQueryConstraints]
-        ] = None,
         ra_field_name: str = "ra",
         dec_field_name: str = "dec",
         order_field_name: Optional[str] = None,
         order_ascending: bool = False,
-        q3c_bool: bool = False,
         query_dist: bool = False,
         **kwargs,
     ):
@@ -219,14 +170,12 @@ def __init__(
         self.xmatch_radius_arcsec = crossmatch_radius_arcsec
         self.ra_field_name = ra_field_name
         self.dec_field_name = dec_field_name
-        self.q3c = q3c_bool
-        self.user_defined_constraints = user_defined_constraints
         self.order_field_name = order_field_name
         self.order_ascending = order_ascending
         self.query_dist = query_dist
 
     def update_dataframe(
-        self, candidate_table: pd.DataFrame, results: list[list[dict]]
+        self, candidate_table: pd.DataFrame, results: list[pd.DataFrame]
     ) -> pd.DataFrame:
         """
         Update a dataframe with db results
@@ -235,23 +184,25 @@ def update_dataframe(
         :param results: results from db query
         :return: updated dataframe
         """
-        assert len(results) == len(candidate_table)
-        keys = results[0][0].keys()
-        for num in range(len(results[0])):
-            for key in keys:
-                candidate_table[f"{key}{num + 1}"] = [x[num][key] for x in results]
-        return candidate_table
+        raise NotImplementedError()
 
-    def get_source_constraints(self, cand: pd.DataFrame) -> None | DBQueryConstraints:
+    def get_source_constraints(self, source: pd.Series) -> DBQueryConstraints:
         """
-        Get db query constraints for a single source
+        Apply constraints to a single source, using q3c
 
-        :param cand: single source
-        :return: db constraint
+        :param source: Source
+        :return: DBQueryConstraints
         """
-        if self.user_defined_constraints is None:
-            return None
-        return self.user_defined_constraints(cand)
+        query_constraints = DBQueryConstraints()
+        query_constraints.add_q3c_constraint(
+            ra=source["ra"],
+            dec=source["dec"],
+            ra_field_name=self.ra_field_name,
+            dec_field_name=self.dec_field_name,
+            crossmatch_radius_arcsec=self.xmatch_radius_arcsec,
+        )
+
+        return query_constraints
 
     def _apply_to_sources(
         self,
@@ -260,30 +211,22 @@ def _apply_to_sources(
         for source_table in batch:
             candidate_table = source_table.get_data()
             results = []
-            for ind in range(len(candidate_table)):
-                cand = candidate_table.loc[ind]
+            for _, cand in candidate_table.iterrows():
                 query_constraints = self.get_source_constraints(cand)
-                res = self.pg_user.crossmatch_with_database(
-                    db_name=self.db_name,
-                    db_table=self.db_table,
-                    db_output_columns=self.db_output_columns,
-                    output_alias_map=self.output_alias_map,
-                    query_constraints=query_constraints,
-                    ra=cand[self.ra_field_name],
-                    dec=cand[self.dec_field_name],
-                    crossmatch_radius_arcsec=self.xmatch_radius_arcsec,
-                    query_distance_bool=self.query_dist,
-                    q3c_bool=self.q3c,
-                    order_field_name=self.order_field_name,
-                    num_limit=self.max_num_results,
+
+                res = select_from_table(
+                    sql_table=self.db_table.sql_model,
+                    db_constraints=query_constraints,
+                    output_columns=self.db_output_columns,
                 )
+
                 results.append(res)
             new_table = self.update_dataframe(candidate_table, results)
             source_table.set_data(new_table)
         return batch
 
 
-class DatabaseHistoryImporter(DatabaseCrossmatchImporter):
+class DatabaseHistorySelector(DatabaseCrossmatchSelector):
     """
     Processor to import previous detections of a source from a database
     """
@@ -292,11 +235,9 @@ def __init__(
         self,
         history_duration_days: float,
         time_field_name: str = "jd",
-        history_key: str = "prv_candidates",
         **kwargs,
     ):
         super().__init__(**kwargs)
-        self.history_key = history_key
         self.history_duration_days = history_duration_days
         self.time_field_name = time_field_name
         logger.info(f"Update db is {self.update_dataframe}")
@@ -304,7 +245,7 @@ def __init__(
     def update_dataframe(
         self,
         candidate_table: pd.DataFrame,
-        results: list[list[dict]],
+        results: list[pd.DataFrame],
     ) -> pd.DataFrame:
         """
         Update a pandas dataframe with the number of previous detections
@@ -314,17 +255,9 @@ def update_dataframe(
         :return: updated pandas dataframe
         """
         assert len(results) == len(candidate_table)
-        candidate_table[self.history_key] = results
-        return candidate_table
 
-    def get_source_constraints(self, cand: pd.DataFrame) -> DBQueryConstraints:
-        t_detection = float(cand[self.time_field_name])
+        candidate_table[SOURCE_HISTORY_KEY] = [
+            x.to_dict(orient="records") for x in results
+        ]
 
-        query_constraints = DBQueryConstraints(
-            columns=self.time_field_name,
-            accepted_values=t_detection - self.history_duration_days,
-            comparison_types=">",
-        )
-        if self.user_defined_constraints is not None:
-            query_constraints += self.user_defined_constraints(cand)
-        return query_constraints
+        return candidate_table
diff --git a/mirar/processors/database/database_updater.py b/mirar/processors/database/database_updater.py
new file mode 100644
index 000000000..adc3bcffe
--- /dev/null
+++ b/mirar/processors/database/database_updater.py
@@ -0,0 +1,137 @@
+"""
+Module for processors to modify database entries
+"""
+import logging
+from abc import ABC
+from typing import Optional
+
+import pandas as pd
+
+from mirar.data import ImageBatch
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.transactions import select_from_table
+from mirar.database.utils import get_sequence_key_names_from_table
+from mirar.processors.database.database_inserter import DatabaseImageInserter
+from mirar.processors.database.database_selector import (
+    BaseDatabaseSelector,
+    BaseImageDatabaseSelector,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class BaseDatabaseUpdater(BaseDatabaseSelector, ABC):
+    """
+    Base class for database updaters
+    """
+
+    base_key = "dbupdater"
+
+    def __init__(self, db_alter_columns: str | list[str], **kwargs):
+        super().__init__(db_output_columns=db_alter_columns, **kwargs)
+        if not isinstance(db_alter_columns, list):
+            db_alter_columns = [db_alter_columns]
+        self.db_alter_columns = db_alter_columns
+
+
+class ImageDatabaseUpdater(BaseDatabaseUpdater, BaseImageDatabaseSelector, ABC):
+    """Base Class for updating image entries in a database"""
+
+    def _apply_to_images(
+        self,
+        batch: ImageBatch,
+    ) -> ImageBatch:
+        for image in batch:
+            val_dict = self.generate_value_dict(image)
+            new = self.db_table(**val_dict)
+            new.update_entry(update_keys=self.db_alter_columns)
+        return batch
+
+    @staticmethod
+    def generate_value_dict(image):
+        """
+        Get the value dictionary for an image
+
+        :param image: Image
+        :return: Value dictionary
+        """
+        return DatabaseImageInserter.generate_value_dict(image)
+
+
+class ImageSequenceDatabaseUpdater(ImageDatabaseUpdater):
+    """
+    Processor to modify images in a database with a sequence
+    """
+
+    def __init__(self, sequence_key: Optional[str | list[str]] = None, **kwargs):
+        super().__init__(**kwargs)
+        self.sequence_key = sequence_key
+
+    def get_constraints(self, data) -> DBQueryConstraints:
+        """
+        Function to get the constraints for a database query
+
+        :param data: Image
+        :return: Constraints for a database query
+        """
+        if self.sequence_key is None:
+            self.sequence_key = get_sequence_key_names_from_table(
+                self.db_table.sql_model.__tablename__, self.db_name
+            )
+
+        accepted_values = [data[x.lower()] for x in self.sequence_key]
+        comparison_types = ["="] * len(accepted_values)
+
+        query_constraints = DBQueryConstraints(
+            columns=self.sequence_key,
+            accepted_values=accepted_values,
+            comparison_types=comparison_types,
+        )
+
+        return query_constraints
+
+
+class ImageDatabaseMultiEntryUpdater(ImageSequenceDatabaseUpdater):
+    """
+    Processor to modify multiple entries specified by a list of sequences in an
+    image database
+    """
+
+    def __init__(self, sequence_key: str, **kwargs):
+        super().__init__(**kwargs)
+        self.sequence_key = sequence_key.lower()
+
+    def _apply_to_images(
+        self,
+        batch: ImageBatch,
+    ) -> ImageBatch:
+        for image in batch:
+            try:
+                unique_key_vals = [int(y) for y in image[self.sequence_key].split(",")]
+            except ValueError as exc:
+                raise ValueError("Sequence keys must be integers") from exc
+
+            data_df = pd.DataFrame(unique_key_vals, columns=[self.sequence_key])
+
+            for _, row in data_df.iterrows():
+                constraints = DBQueryConstraints(
+                    columns=self.sequence_key, accepted_values=row[self.sequence_key]
+                )
+
+                old = select_from_table(
+                    db_constraints=constraints,
+                    sql_table=self.db_table.sql_model,
+                )
+
+                assert (
+                    len(old) == 1
+                ), f"Multiple entries found for unique key {self.sequence_key}"
+
+                for key in self.db_alter_columns:
+                    old[key] = image[key]
+
+                new = self.db_table(**old.to_dict(orient="records")[0])
+
+                new.update_entry(update_keys=self.db_alter_columns)
+
+        return batch
diff --git a/mirar/processors/database/postgres.py b/mirar/processors/database/postgres.py
deleted file mode 100644
index a856d1ad3..000000000
--- a/mirar/processors/database/postgres.py
+++ /dev/null
@@ -1,691 +0,0 @@
-"""
-Module containing postgres util functions
-"""
-# pylint: disable=not-context-manager
-import logging
-import os
-from glob import glob
-from pathlib import Path
-from typing import Optional
-
-import numpy as np
-import psycopg
-from psycopg import errors
-from psycopg.rows import Row
-
-from mirar.data import DataBlock
-from mirar.errors import ProcessorError
-from mirar.processors.database.constraints import DBQueryConstraints
-from mirar.processors.database.utils import get_ordered_schema_list
-
-logger = logging.getLogger(__name__)
-
-DB_USER_KEY = "DB_USER"
-DB_PASSWORD_KEY = "DB_PWD"
-
-PG_ADMIN_USER_KEY = "PG_ADMIN_USER"
-PG_ADMIN_PWD_KEY = "PG_ADMIN_PWD"
-
-DB_USER = os.getenv(DB_USER_KEY)
-DB_PASSWORD = os.getenv(DB_PASSWORD_KEY)
-
-ADMIN_USER = os.getenv(PG_ADMIN_USER_KEY, DB_USER)
-ADMIN_PASSWORD = os.getenv(PG_ADMIN_PWD_KEY, DB_PASSWORD)
-
-POSTGRES_DUPLICATE_PROTOCOLS = ["fail", "ignore", "replace"]
-
-
-class DataBaseError(ProcessorError):
-    """Error relating to postgres interactions"""
-
-
-class PostgresUser:
-    """
-    Basic Postgres user class for executing functions
-    """
-
-    user_env_varaiable = DB_USER_KEY
-    pass_env_variable = DB_PASSWORD_KEY
-
-    def __init__(self, db_user: str = DB_USER, db_password: str = DB_PASSWORD):
-        self.db_user = db_user
-        self.db_password = db_password
-
-    def validate_credentials(self):
-        """
-        Checks that user credentials exist
-        :return: None
-        """
-        if self.db_user is None:
-            err = (
-                f"'db_user' is set as None. Please pass a db_user as an argument, "
-                f"or set the environment variable '{self.user_env_varaiable}'."
-            )
-            logger.error(err)
-            raise DataBaseError(err)
-
-        if self.db_password is None:
-            err = (
-                f"'db_password' is set as None. Please pass a password as an argument, "
-                f"or set the environment variable '{self.pass_env_variable}'."
-            )
-            logger.error(err)
-            raise DataBaseError(err)
-
-        # TODO check user exists
-
-    def run_sql_command_from_file(self, file_path: str | Path, db_name: str):
-        """
-        Execute SQL command from file
-
-        :param file_path: File to execute
-        :param db_name: name of database
-        :return: False
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            with open(file_path, "r", encoding="utf8") as sql_file:
-                conn.execute(sql_file.read())
-
-            logger.info(f"Executed sql commands from file {file_path}")
-
-    def create_table(self, schema_path: str | Path, db_name: str):
-        """
-        Create a database table
-
-        :param schema_path: File to execute
-        :param db_name: name of database
-        :return: None
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            with open(schema_path, "r", encoding="utf8") as schema_file:
-                conn.execute(schema_file.read())
-
-        logger.info(f"Created table from schema path {schema_path}")
-
-    def create_tables_from_schema(
-        self,
-        schema_dir: str | Path,
-        db_name: str,
-    ):
-        """
-        Creates a db with tables, as described by .sql files in a directory
-
-        :param schema_dir: Directory containing schema files
-        :param db_name: name of DB
-        :return: None
-        """
-        schema_files = glob(f"{schema_dir}/*.sql")
-        ordered_schema_files = get_ordered_schema_list(schema_files)
-        logger.info(f"Creating the following tables - {ordered_schema_files}")
-        for schema_file in ordered_schema_files:
-            self.create_table(schema_path=schema_file, db_name=db_name)
-
-    def export_to_db(
-        self,
-        value_dict: dict | DataBlock,
-        db_name: str,
-        db_table: str,
-        duplicate_protocol: str = "fail",
-    ) -> tuple[list, list]:
-        """
-        Export a list of fields in value dict to a batabase table
-
-        :param value_dict: dictionary/DataBlock/other dictonary-like object to export
-        :param db_name: name of db to export to
-        :param db_table: table of DB to export to
-        :param duplicate_protocol: protocol for handling duplicates,
-            in "fail"/"ignore"/"replace"
-        :return:
-        """
-
-        assert duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS
-
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-
-            sql_query = f"""
-            SELECT Col.Column_Name from
-                INFORMATION_SCHEMA.TABLE_CONSTRAINTS Tab,
-                INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE Col
-            WHERE
-                Col.Constraint_Name = Tab.Constraint_Name
-                AND Col.Table_Name = Tab.Table_Name
-                AND Constraint_Type = 'PRIMARY KEY'
-                AND Col.Table_Name = '{db_table}'
-            """
-            serial_keys, serial_key_values = [], []
-            with conn.execute(sql_query) as cursor:
-                primary_key = [x[0] for x in cursor.fetchall()]
-                serial_keys = list(self.get_sequence_keys_from_table(db_table, db_name))
-                logger.debug(serial_keys)
-                colnames = [
-                    desc[0]
-                    for desc in conn.execute(
-                        f"SELECT * FROM {db_table} LIMIT 1"
-                    ).description
-                    if desc[0] not in serial_keys
-                ]
-
-                colnames_str = ""
-                for column in colnames:
-                    colnames_str += f'"{column}",'
-                colnames_str = colnames_str[:-1]
-                txt = f"INSERT INTO {db_table} ({colnames_str}) VALUES ("
-
-                for char in ["[", "]", "'"]:
-                    txt = txt.replace(char, "")
-
-                for column in colnames:
-                    txt += f"'{str(value_dict[column])}', "
-
-                txt = txt + ") "
-                txt = txt.replace(", )", ")")
-
-                if len(serial_keys) > 0:
-                    txt += "RETURNING "
-                    for key in serial_keys:
-                        txt += f"{key},"
-                    txt += ";"
-                    txt = txt.replace(",;", ";")
-
-                logger.debug(txt)
-                command = txt
-
-                try:
-                    cursor.execute(command)
-                    if len(serial_keys) > 0:
-                        serial_key_values = cursor.fetchall()[0]
-                    else:
-                        serial_key_values = []
-
-                except errors.UniqueViolation as exc:
-                    primary_key_values = [value_dict[x] for x in primary_key]
-
-                    if duplicate_protocol == "fail":
-                        err = (
-                            f"Duplicate error, entry with "
-                            f"{primary_key}={primary_key_values} "
-                            f"already exists in {db_name}."
-                        )
-                        logger.error(err)
-                        raise errors.UniqueViolation from exc
-
-                    if duplicate_protocol == "ignore":
-                        logger.debug(
-                            f"Found duplicate entry with "
-                            f"{primary_key}={primary_key_values} in {db_name}. "
-                            f"Ignoring."
-                        )
-                    elif duplicate_protocol == "replace":
-                        logger.debug(
-                            f"Updating duplicate entry with "
-                            f"{primary_key}={primary_key_values} in {db_name}."
-                        )
-
-                        db_constraints = DBQueryConstraints(
-                            columns=primary_key,
-                            accepted_values=primary_key_values,
-                        )
-
-                        update_colnames = []
-                        for column in colnames:
-                            if column not in primary_key:
-                                update_colnames.append(column)
-
-                        if len(serial_keys) == 0:
-                            self.modify_db_entry(
-                                db_constraints=db_constraints,
-                                value_dict=value_dict,
-                                db_alter_columns=update_colnames,
-                                db_table=db_table,
-                                db_name=db_name,
-                                return_columns=primary_key,
-                            )
-
-                        else:
-                            serial_key_values = self.modify_db_entry(
-                                db_constraints=db_constraints,
-                                value_dict=value_dict,
-                                db_alter_columns=update_colnames,
-                                db_table=db_table,
-                                db_name=db_name,
-                                return_columns=primary_key,
-                            )
-
-        return serial_keys, serial_key_values
-
-    def modify_db_entry(
-        self,
-        db_name: str,
-        db_table: str,
-        db_constraints: DBQueryConstraints,
-        value_dict: dict | DataBlock,
-        db_alter_columns: str | list[str],
-        return_columns: Optional[str | list[str]] = None,
-    ) -> list[Row]:
-        """
-        Modify a db entry
-
-        :param db_name: name of db
-        :param db_table: Name of table
-        :param db_constraints: constraints to query db
-        :param value_dict: dict-like object to provide updated values
-        :param db_alter_columns: columns to alter in db
-        :param return_columns: columns to return
-        :return: db query (return columns)
-        """
-
-        if not isinstance(db_alter_columns, list):
-            db_alter_columns = [db_alter_columns]
-
-        if return_columns is None:
-            return_columns = db_alter_columns
-        if not isinstance(return_columns, list):
-            return_columns = [return_columns]
-
-        constraints = db_constraints.parse_constraints()
-
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-
-            db_alter_values = [str(value_dict[c]) for c in db_alter_columns]
-
-            alter_values_txt = [
-                f'"{db_alter_columns[ind]}"=' + f"'{db_alter_values[ind]}'"
-                for ind in range(len(db_alter_columns))
-            ]
-
-            sql_query = (
-                f"UPDATE {db_table} SET {', '.join(alter_values_txt)} "
-                f"WHERE {constraints}"
-            )
-
-            if len(return_columns) > 0:
-                logger.debug(return_columns)
-                sql_query += f""" RETURNING {', '.join(return_columns)}"""
-
-            sql_query += ";"
-            query_output = self.execute_query(sql_query, db_name)
-
-        return query_output
-
-    def get_sequence_keys_from_table(self, db_table: str, db_name: str) -> np.ndarray:
-        """
-        Gets sequence keys of db table
-
-        :param db_table: database table to use
-        :param db_name: dataname name
-        :return: numpy array of keys
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            sequences = [
-                x[0]
-                for x in conn.execute(
-                    "SELECT c.relname FROM pg_class c WHERE c.relkind = 'S';"
-                ).fetchall()
-            ]
-            seq_tables = np.array([x.split("_")[0] for x in sequences])
-            seq_columns = np.array([x.split("_")[1] for x in sequences])
-            table_sequence_keys = seq_columns[(seq_tables == db_table)]
-        return table_sequence_keys
-
-    def import_from_db(
-        self,
-        db_name: str,
-        db_table: str,
-        db_output_columns: str | list[str],
-        output_alias_map: Optional[str | list[str]] = None,
-        max_num_results: Optional[int] = None,
-        db_constraints: Optional[DBQueryConstraints] = None,
-    ) -> list[dict]:
-        """Query an SQL database with constraints, and return a list of dictionaries.
-        One dictionary per entry returned from the query.
-
-        #TODO check admin
-
-        Parameters
-        ----------
-        db_name: Name of database to query
-        db_table: Name of database table to query
-        db_output_columns: Name(s) of columns to return for matched database entries
-        output_alias_map: Alias to assign for each output column
-        max_num_results: Maximum number of results to return
-
-        Returns
-        -------
-        A list of dictionaries (one per entry)
-        """
-
-        if not isinstance(db_output_columns, list):
-            db_output_columns = [db_output_columns]
-
-        if output_alias_map is None:
-            output_alias_map = db_output_columns
-
-        if not isinstance(output_alias_map, list):
-            output_alias_map = [output_alias_map]
-
-        assert len(output_alias_map) == len(db_output_columns)
-
-        all_query_res = []
-
-        if db_constraints is not None:
-            constraints = db_constraints.parse_constraints()
-        else:
-            constraints = ""
-
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            sql_query = f"""
-            SELECT {', '.join(db_output_columns)} from {db_table}
-                WHERE {constraints}
-            """
-
-            if max_num_results is not None:
-                sql_query += f" LIMIT {max_num_results}"
-
-            sql_query += ";"
-
-            logger.debug(f"Query: {sql_query}")
-
-            with conn.execute(sql_query) as cursor:
-                query_output = cursor.fetchall()
-
-            for entry in query_output:
-                assert len(entry) == len(db_output_columns)
-
-                query_res = {}
-
-                for i, key in enumerate(output_alias_map):
-                    query_res[key] = entry[i]
-
-                all_query_res.append(query_res)
-
-        return all_query_res
-
-    def execute_query(self, sql_query: str, db_name: str) -> list[Row]:
-        """
-        Generically execute SQL query
-
-        :param sql_query: SQL query to execute
-        :param db_name: db name
-        :return: rows from db
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            logger.debug(f"Query: {sql_query}")
-
-            with conn.execute(sql_query) as cursor:
-                query_output = cursor.fetchall()
-
-        return query_output
-
-    def crossmatch_with_database(
-        self,
-        db_name: str,
-        db_table: str,
-        db_output_columns: str | list[str],
-        ra: float,
-        dec: float,
-        crossmatch_radius_arcsec: float,
-        output_alias_map: Optional[dict] = None,
-        ra_field_name: str = "ra",
-        dec_field_name: str = "dec",
-        query_distance_bool: bool = False,
-        q3c_bool: bool = False,
-        query_constraints: Optional[DBQueryConstraints] = None,
-        order_field_name: Optional[str] = None,
-        num_limit: Optional[int] = None,
-    ) -> list[dict]:
-        """
-        Crossmatch a given spatial position (ra/dec) with sources in a database,
-        and returns a list of matches
-
-        #TODO: check admin
-
-        :param db_name: name of db to query
-        :param db_table: name of db table
-        :param db_output_columns: columns to return
-        :param output_alias_map: mapping for renaming columns
-        :param ra: RA
-        :param dec: dec
-        :param crossmatch_radius_arcsec: radius for crossmatch
-        :param ra_field_name: name of ra column in database
-        :param dec_field_name: name of dec column in database
-        :param query_distance_bool: boolean where to return crossmatch distance
-        :param q3c_bool: boolean whether to use q3c_bool
-        :param order_field_name: field to order result by
-        :param num_limit: limit on sql query
-        :return: list of query result dictionaries
-        """
-
-        if output_alias_map is None:
-            output_alias_map = {}
-            for col in db_output_columns:
-                output_alias_map[col] = col
-
-        crossmatch_radius_deg = crossmatch_radius_arcsec / 3600.0
-
-        if q3c_bool:
-            constraints = (
-                f"q3c_radial_query({ra_field_name},{dec_field_name},"
-                f"{ra},{dec},{crossmatch_radius_deg}) "
-            )
-        else:
-            ra_min = ra - crossmatch_radius_deg
-            ra_max = ra + crossmatch_radius_deg
-            dec_min = dec - crossmatch_radius_deg
-            dec_max = dec + crossmatch_radius_deg
-            constraints = (
-                f" {ra_field_name} between {ra_min} and {ra_max} AND "
-                f"{dec_field_name} between {dec_min} and {dec_max} "
-            )
-
-        if query_constraints is not None:
-            constraints += f"""AND {query_constraints.parse_constraints()}"""
-
-        select = f""" {'"' + '","'.join(db_output_columns) + '"'}"""
-        if query_distance_bool:
-            if q3c_bool:
-                select = (
-                    f"q3c_dist({ra_field_name},{dec_field_name},{ra},{dec}) AS xdist,"
-                    + select
-                )
-            else:
-                select = f"""{ra_field_name} - ra AS xdist,""" + select
-
-        query = f"""SELECT {select} FROM {db_table} WHERE {constraints}"""
-
-        if order_field_name is not None:
-            query += f""" ORDER BY {order_field_name}"""
-        if num_limit is not None:
-            query += f""" LIMIT {num_limit}"""
-
-        query += ";"
-
-        query_output = self.execute_query(query, db_name)
-        all_query_res = []
-
-        for entry in query_output:
-            if not query_distance_bool:
-                assert len(entry) == len(db_output_columns)
-            else:
-                assert len(entry) == len(db_output_columns) + 1
-            query_res = {}
-            for i, key in enumerate(output_alias_map):
-                query_res[key] = entry[i]
-                if query_distance_bool:
-                    query_res["dist"] = entry["xdist"]
-            all_query_res.append(query_res)
-        return all_query_res
-
-    def check_if_exists(
-        self, check_command: str, check_value: str, db_name: str = "postgres"
-    ) -> bool:
-        """
-        Check if a user account exists
-
-        :param check_command if a user/database/table exists
-        :param check_value: username to check
-        :param db_name: name of database to query
-        :return: boolean
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            data = conn.execute(check_command).fetchall()
-        existing_user_names = [x[0] for x in data]
-        logger.debug(f"Found the following values: {existing_user_names}")
-
-        return check_value in existing_user_names
-
-    def create_db(self, db_name: str):
-        """
-        Creates a database using credentials
-
-        :param db_name: DB to create
-        :return: None
-        """
-
-        with psycopg.connect(
-            f"dbname=postgres user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            sql = f"""CREATE database {db_name}"""
-            conn.execute(sql)
-            logger.info(f"Created db {db_name}")
-
-    def check_if_db_exists(self, db_name: str) -> bool:
-        """
-        Check if a user account exists
-
-        :param db_name: database to check
-        :return: boolean
-        """
-
-        check_command = """SELECT datname FROM pg_database;"""
-
-        db_exist_bool = self.check_if_exists(
-            check_command=check_command,
-            check_value=db_name,
-            db_name="postgres",
-        )
-
-        logger.debug(f"Database '{db_name}' does {['not ', ''][db_exist_bool]} exist")
-
-        return db_exist_bool
-
-    def check_if_table_exists(self, db_name: str, db_table: str) -> bool:
-        """
-        Check if a db table account exists
-
-        :param db_name: database to check
-        :param db_table: table to check
-        :return: boolean
-        """
-
-        check_command = (
-            "SELECT table_name FROM information_schema.tables "
-            "WHERE table_schema='public';"
-        )
-
-        table_exist_bool = self.check_if_exists(
-            check_command=check_command,
-            check_value=db_table,
-            db_name=db_name,
-        )
-
-        logger.debug(f"Table '{db_table}' does {['not ', ''][table_exist_bool]} exist")
-
-        return table_exist_bool
-
-
-class PostgresAdmin(PostgresUser):
-    """
-    An Admin postgres user, with additional functionality for creatying new users
-    """
-
-    user_env_varaiable = PG_ADMIN_USER_KEY
-    pass_env_variable = PG_ADMIN_PWD_KEY
-
-    def __init__(self, db_user: str = ADMIN_USER, db_password: str = ADMIN_PASSWORD):
-        super().__init__(db_user=db_user, db_password=db_password)
-
-    def create_new_user(self, new_db_user: str, new_password: str):
-        """
-        Create a new postgres user
-
-        :param new_db_user: new username
-        :param new_password: new user password
-        :return: None
-        """
-
-        with psycopg.connect(
-            f"dbname=postgres user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            command = f"CREATE ROLE {new_db_user} WITH password '{new_password}' LOGIN;"
-            conn.execute(command)
-
-    def grant_privileges(self, db_name: str, db_user: str, schema: str = "public"):
-        """
-        Grant privilege to user on database
-
-        :param db_name: name of database
-        :param db_user: username to grant privileges for db_user
-        :return: None
-        """
-        with psycopg.connect(
-            f"dbname=postgres user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            command = f"""GRANT ALL PRIVILEGES ON DATABASE {db_name} TO {db_user};"""
-            conn.execute(command)
-            command = f"""ALTER DATABASE {db_name} OWNER TO {db_user}"""
-            conn.execute(command)
-
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            command = f"""GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA {schema}
-            TO {db_user};"""
-            conn.execute(command)
-            command = f"""GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA {schema}
-                        TO {db_user};"""
-            conn.execute(command)
-
-    def check_if_user_exists(self, user_name: str) -> bool:
-        """
-        Check if a user account exists
-
-        :param user_name: username to check
-        :return: boolean
-        """
-        check_command = """SELECT usename FROM pg_user;"""
-
-        user_exist_bool = self.check_if_exists(
-            check_command=check_command,
-            check_value=user_name,
-        )
-
-        logger.debug(f"User '{user_name}' does {['not ', ''][user_exist_bool]} exist")
-
-        return user_exist_bool
diff --git a/mirar/processors/database/utils.py b/mirar/processors/database/utils.py
deleted file mode 100644
index e606066d6..000000000
--- a/mirar/processors/database/utils.py
+++ /dev/null
@@ -1,80 +0,0 @@
-"""
-Module containing postgres util functions
-"""
-from pathlib import Path
-
-import numpy as np
-
-
-def get_column_names_from_schema(schema_file_path: str | Path) -> list[str]:
-    """
-    Get column names from a schema file
-
-    :param schema_file_path: file to read
-    :return: list of columns
-    """
-    with open(schema_file_path, "r", encoding="utf8") as schema_file:
-        dat = schema_file.read()
-    dat = dat.split(");")[0]
-    dat = dat.split("\n")[1:-1]
-    pkstrip = [x.strip(",").split("PRIMARY KEY")[0].strip() for x in dat]
-    fkstrip = [x.strip(",").split("FOREIGN KEY")[0].strip() for x in pkstrip]
-    colnames = [x.split(" ")[0].strip('"') for x in fkstrip]
-    return colnames
-
-
-def get_foreign_tables_list(schema_files: list[str]) -> np.ndarray:
-    """
-    Returns a list of foreign tables
-
-    :param schema_files: List of schema files to read
-    :return: Returns list of foreign keys in schema
-    """
-    foreign_tables_list = []
-    for schema_file_path in schema_files:
-        table_names = []
-        with open(schema_file_path, "r", encoding="utf8") as schema_file:
-            schema = schema_file.read()
-        if "FOREIGN KEY" in schema:
-            schema = schema.replace("\n", "")
-            schema = schema.replace("\t", "")
-            schema_split = np.array(schema.split(","))
-            fk_rows = np.array(["FOREIGN KEY" in x for x in schema_split])
-            for row in schema_split[fk_rows]:
-                words = np.array(row.split(" "))
-                refmask = np.array(["REFERENCES" in x for x in words])
-                idx = np.where(refmask)[0][0] + 1
-                tablename = words[idx].split("(")[0]
-                table_names.append(tablename)
-        foreign_tables_list.append(np.array(table_names))
-    return np.array(foreign_tables_list)
-
-
-def get_ordered_schema_list(schema_files: list[str]) -> list[str]:
-    """
-    Returns an ordered list of schema, ensuring connected database tables
-    are created in the right order
-
-    :param schema_files: List of schema files to read
-    :return: Returns list of foreign keys in schema
-    """
-    foreign_tables_list = get_foreign_tables_list(schema_files)
-    ordered_schema_list = []
-    tables_created = []
-    schema_table_names = [x.split("/")[-1].split(".sql")[0] for x in schema_files]
-    while len(tables_created) < len(schema_files):
-        for ind, schema_file in enumerate(schema_files):
-            table_name = schema_table_names[ind]
-            if table_name in tables_created:
-                pass
-            else:
-                foreign_tables = foreign_tables_list[ind]
-                if len(foreign_tables) == 0:
-                    ordered_schema_list.append(schema_file)
-                    tables_created.append(table_name)
-                else:
-                    if np.all(np.isin(foreign_tables, tables_created)):
-                        ordered_schema_list.append(schema_file)
-                        tables_created.append(table_name)
-
-    return ordered_schema_list
diff --git a/mirar/processors/photometry/utils.py b/mirar/processors/photometry/utils.py
index d64292b84..b35195dcb 100644
--- a/mirar/processors/photometry/utils.py
+++ b/mirar/processors/photometry/utils.py
@@ -251,10 +251,13 @@ def aper_photometry(
     # error = calc_total_error(data, bkg_error, effective_gain)
     # phot_table = aperture_photometry(diff_cutout - bkg, aperture, error=error)
     # counts_err = phot_table['aperture_sum_err'][0]
-    error = np.sqrt(np.sum(aperture_unc_data**2))
+    error = np.sqrt(np.nansum(aperture_unc_data**2.0))
     phot_table = aperture_photometry(image_cutout - bkg, aperture)
     counts = phot_table["aperture_sum"][0]
     counts_err = error
+
+    assert counts_err > 0.0
+
     return counts, counts_err
 
 
diff --git a/mirar/processors/sources/namer.py b/mirar/processors/sources/namer.py
index 28ba32e0f..f4e8dc1e7 100644
--- a/mirar/processors/sources/namer.py
+++ b/mirar/processors/sources/namer.py
@@ -3,17 +3,20 @@
 """
 import logging
 
+import numpy as np
 from astropy.time import Time
+from sqlalchemy import select, text
 
 from mirar.data import SourceBatch
-from mirar.paths import CAND_NAME_KEY
-from mirar.processors.base_processor import BaseSourceProcessor
-from mirar.processors.database import BaseDatabaseProcessor
+from mirar.database.transactions.select import run_select
+from mirar.paths import CAND_NAME_KEY, SOURCE_HISTORY_KEY
+from mirar.processors.database import DatabaseHistorySelector
+from mirar.processors.database.database_selector import DatabaseSourceSelector
 
 logger = logging.getLogger(__name__)
 
 
-class CandidateNamer(BaseDatabaseProcessor, BaseSourceProcessor):
+class CandidateNamer(DatabaseSourceSelector):
     """Processor to sequentially assign names to sources, of the form a, aa, aba..."""
 
     base_key = "namer"
@@ -31,7 +34,7 @@ def __init__(
         date_field: str = "jd",
         **kwargs,
     ):
-        super().__init__(**kwargs)
+        super().__init__(db_output_columns=[db_name_field], **kwargs)
         self.db_name_field = db_name_field
         self.db_order_field = db_order_field
         self.base_name = base_name
@@ -86,18 +89,18 @@ def get_next_name(self, cand_jd: float, last_name: str = None) -> str:
         """
         cand_year = Time(cand_jd, format="jd").datetime.year % 1000
         if last_name is None:
-            query = (
-                f'SELECT "{self.db_name_field}" FROM {self.db_table} '
-                f"ORDER BY {self.db_order_field} desc LIMIT 1;"
+            res = run_select(
+                query=select(getattr(self.db_table.sql_model, self.db_name_field))
+                .order_by(text(f"{self.db_order_field} desc"))
+                .limit(1),
+                sql_table=self.db_table.sql_model,
             )
 
-            res = self.pg_user.execute_query(query, db_name=self.db_name)
-
             if len(res) == 0:
                 name = self.base_name + str(cand_year) + self.name_start
                 return name
 
-            last_name = res[0][0]
+            last_name = res[CAND_NAME_KEY].iloc[0]
             logger.debug(res)
 
         last_year = int(last_name[len(self.base_name) : len(self.base_name) + 2])
@@ -110,54 +113,46 @@ def get_next_name(self, cand_jd: float, last_name: str = None) -> str:
         logger.debug(f"Assigning name: {name}")
         return name
 
-    def is_detected_previously(self, ra_deg: float, dec_deg: float) -> tuple[bool, str]:
-        """
-        Checks whether a source has been detected previously
-
-        :param ra_deg: ra (deg)
-        :param dec_deg: dec (deg)
-        :return: boolean whether a source has been detected previously
-        """
-        name = self.pg_user.crossmatch_with_database(
-            db_name=self.db_name,
-            db_table=self.db_table,
-            db_output_columns=[self.db_name_field],
-            num_limit=1,
-            ra=ra_deg,
-            dec=dec_deg,
-            crossmatch_radius_arcsec=self.crossmatch_radius_arcsec,
-        )  # [0]
-        return len(name) > 0, name
-
     def _apply_to_sources(
         self,
         batch: SourceBatch,
     ) -> SourceBatch:
         for source_table in batch:
-            candidate_table = source_table.get_data()
+            sources = source_table.get_data()
+
+            assert (
+                SOURCE_HISTORY_KEY in sources.columns
+            ), "No candidate history in source table"
 
             names = []
             lastname = None
-            for ind in range(len(candidate_table)):
-                cand = candidate_table.loc[ind]
-                if len(cand["prv_candidates"]) > 0:
-                    cand_name = cand["prv_candidates"][0][self.db_name_field]
+            for _, source in sources.iterrows():
+                if len(source[SOURCE_HISTORY_KEY]) > 0:
+                    source_name = source[SOURCE_HISTORY_KEY][0][self.db_name_field]
 
                 else:
-                    prv_det, prv_name = self.is_detected_previously(
-                        cand["ra"], cand["dec"]
+                    source_name = self.get_next_name(
+                        source[self.date_field], last_name=lastname
                     )
+                    lastname = source_name
+                names.append(source_name)
 
-                    if prv_det:
-                        cand_name = prv_name
-                    else:
-                        cand_name = self.get_next_name(
-                            cand[self.date_field], last_name=lastname
-                        )
-                    lastname = cand_name
-                names.append(cand_name)
-
-            candidate_table[self.db_name_field] = names
-            source_table.set_data(candidate_table)
+            sources[self.db_name_field] = names
+            source_table.set_data(sources)
 
         return batch
+
+    def check_prerequisites(
+        self,
+    ):
+        check = np.sum(
+            [isinstance(x, DatabaseHistorySelector) for x in self.preceding_steps]
+        )
+        if check < 1:
+            err = (
+                f"{self.__module__} requires {DatabaseHistorySelector} "
+                f"as a prerequisite. "
+                f"However, the following steps were found: {self.preceding_steps}."
+            )
+            logger.error(err)
+            raise ValueError(err)
diff --git a/mirar/processors/sqldatabase/__init__.py b/mirar/processors/sqldatabase/__init__.py
deleted file mode 100644
index 165858de2..000000000
--- a/mirar/processors/sqldatabase/__init__.py
+++ /dev/null
@@ -1,9 +0,0 @@
-"""
-Central module for interactions with postgres databases
-"""
-from mirar.processors.database.base_database_processor import BaseDatabaseProcessor
-from mirar.processors.database.postgres import (
-    POSTGRES_DUPLICATE_PROTOCOLS,
-    PostgresAdmin,
-    PostgresUser,
-)
diff --git a/mirar/processors/sqldatabase/base_database_processor.py b/mirar/processors/sqldatabase/base_database_processor.py
deleted file mode 100644
index b4d346a68..000000000
--- a/mirar/processors/sqldatabase/base_database_processor.py
+++ /dev/null
@@ -1,42 +0,0 @@
-"""
-Module containing base database processor class
-"""
-import logging
-from abc import ABC
-from typing import Type
-
-from mirar.processors.base_processor import BaseProcessor
-from mirar.processors.sqldatabase.base_model import BaseDB
-from mirar.processors.sqldatabase.postgres import (
-    POSTGRES_DUPLICATE_PROTOCOLS,
-    PostgresAdmin,
-    PostgresUser,
-)
-
-logger = logging.getLogger(__name__)
-
-
-class BaseDatabaseProcessor(BaseProcessor, ABC):
-    """Base class for processors which interact with a postgres database"""
-
-    max_n_cpu = 1
-
-    def __init__(
-        self,
-        db_table: Type[BaseDB],
-        pg_user: PostgresUser = PostgresUser(),
-        pg_admin: PostgresAdmin = PostgresAdmin(),
-        duplicate_protocol: str = "fail",
-        q3c_bool: bool = False,
-    ):
-        super().__init__()
-        self.db_table = db_table
-        self.db_name = self.db_table.sql_model.db_name
-        self.db_check_bool = False
-        self.duplicate_protocol = duplicate_protocol
-        self.q3c = q3c_bool
-
-        self.pg_user = pg_user
-        self._pg_admin = pg_admin
-
-        assert self.duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS
diff --git a/mirar/processors/sqldatabase/base_model.py b/mirar/processors/sqldatabase/base_model.py
deleted file mode 100644
index 6fbee04ae..000000000
--- a/mirar/processors/sqldatabase/base_model.py
+++ /dev/null
@@ -1,356 +0,0 @@
-"""
-Module to define PSQL database tables using sqlalchemy
-"""
-import logging
-from datetime import date
-from typing import ClassVar
-
-import numpy as np
-from pydantic import BaseModel, Extra, Field, root_validator, validator
-from sqlalchemy import Insert, Select, Table, Update, and_, inspect, select
-
-from mirar.processors.sqldatabase.postgres_utils import (
-    get_sequence_key_names_from_table,
-)
-from mirar.utils.sql import get_engine
-
-logger = logging.getLogger(__name__)
-
-
-class PydanticBase(BaseModel):
-    """
-    Base code pydantic model (no extra colunns!)
-    """
-
-    class Config:
-        """
-        Config
-        """
-
-        extra = Extra.forbid
-
-
-class BaseDB(PydanticBase):
-    """
-    Base Database Table model, requiring an associated SQLalchemy table
-    """
-
-    sql_model: ClassVar[Table]
-
-    @root_validator(pre=True)
-    def sql_model_exists_check(cls, values):
-        """
-        Validator to ensure an sql model has been specified in the child class
-        :param values: values
-        :return: values
-        """
-
-        if "sql_model" not in cls.__annotations__:
-            raise ValueError("sql model must be set in class")
-        return values
-
-    @validator("*")
-    @classmethod
-    def validate_sql(cls, value, field):
-        """
-        Validator to ensure that the field names of a pydantic model
-        match the database table
-        :param value: value
-        :param field: field
-        :return: value
-        """
-        if field.name not in cls.sql_model.__table__.columns.keys():
-            err = f"Field '{field.name}' not duplicated in {cls.sql_model}"
-            raise ValueError(err)
-        return value
-
-    def _insert_entry(self, returning_key_names: list = None) -> tuple:
-        """
-        Insert the pydantic-ified data into the corresponding sql database
-        :return: sequence_key_names, sequence_key_values of the updated entry
-        """
-        if returning_key_names is None:
-            returning_keys = self.get_sequence_keys()
-        else:
-            if not isinstance(returning_key_names, list):
-                returning_key_names = [returning_key_names]
-
-            returning_keys = self.get_table_keys_from_names(returning_key_names)
-
-        logger.debug(f"Returning keys {returning_keys}")
-        stmt = Insert(self.sql_model).values(**self.dict()).returning(*returning_keys)
-        engine = get_engine(db_name=self.sql_model.db_name)
-        with engine.connect() as conn:
-            res = conn.execute(stmt)
-            conn.commit()
-
-        if len(returning_keys) == 0:
-            returning_values = []
-        else:
-            returning_values = res.fetchall()[0]
-        logger.debug(returning_values)
-
-        returning_key_names = [x.key for x in returning_keys]
-        assert len(returning_values) == len(returning_key_names)
-        return returning_key_names, returning_values
-
-    def insert_entry(self, returning_key_names: str | list = None):
-        """
-        Insert the pydantic-ified data into the corresponding sql database
-        :return: sequence_key_names, sequence_key_values of the updated entry
-        """
-        result = self._insert_entry(returning_key_names=returning_key_names)
-        logger.debug(f"Return result {result}")
-        return result
-
-    def get_sequence_keys(self) -> list:
-        """
-        Get sequence column instances of the sql_model table
-        Returns:
-        list of sequence keys
-        """
-        sequence_key_names = get_sequence_key_names_from_table(
-            db_name=self.sql_model.db_name, db_table=self.sql_model.__tablename__
-        )
-        sequence_keys = self.get_table_keys_from_names(key_names=sequence_key_names)
-        return sequence_keys
-
-    def get_table_keys_from_names(self, key_names: list | np.ndarray) -> list:
-        """
-        Function to get column instances from names of sql table columns
-        Args:
-            key_names:
-        Returns:
-            list
-        """
-        return [self.sql_model.__dict__[x] for x in key_names]
-
-    def _update_entry(self, primary_key_val, update_key_names=None) -> tuple:
-        """
-        Update database entry
-        Args:
-            primary_key_val: value of primary key to find the db entry
-            update_key_names: names of keys to be updates, if None, will update all keys
-        Returns:
-            sequence_key_names, sequence_key_values of the updated entry
-        """
-        primary_key = inspect(self.sql_model).primary_key[0]
-
-        if update_key_names is None:
-            update_key_names = [x for x in self.dict() if x != primary_key.name]
-
-        update_keys = self.get_table_keys_from_names(update_key_names)
-        update_vals = [self.dict()[x] for x in update_key_names]
-
-        returning_keys = self.get_sequence_keys()
-
-        update_dict = {}
-        for x, key in enumerate(update_keys):
-            update_dict[key.key] = update_vals[x]
-
-        stmt = (
-            Update(self.sql_model)
-            .values(**update_dict)
-            .where(primary_key == primary_key_val)
-            .returning(*returning_keys)
-        )
-
-        logger.debug(stmt)
-        engine = get_engine(db_name=self.sql_model.db_name)
-        with engine.connect() as conn:
-            res = conn.execute(stmt)
-            conn.commit()
-
-        if len(returning_keys) == 0:
-            returning_values = []
-        else:
-            returning_values = res.fetchall()[0]
-        returning_key_names = [x.key for x in returning_keys]
-        logger.debug(f"{returning_key_names}, {returning_values}")
-        assert len(returning_values) == len(returning_key_names)
-        return returning_key_names, returning_values
-
-    def update_entry(self, primary_key_val, update_keys=None) -> tuple:
-        """
-        Wrapper to update database entry. Users should override this function.
-        Args:
-            primary_key_val:
-            update_keys:
-        Returns:
-            sequence_key_names, sequence_key_values of the updated entry
-        """
-        return self._update_entry(primary_key_val, update_keys)
-
-
-class BaseTable:
-    """
-    Parent class for database tables. Tables should inherit from this
-    and DeclarativeBase.
-    """
-
-    @property
-    def __tablename__(self):
-        raise NotImplementedError
-
-    @property
-    def db_name(self):
-        """
-        Name of the database.
-        :return: None
-        """
-        raise NotImplementedError
-
-    def get_primary_key(self) -> str:
-        """
-        Function to get primary key of table
-        Returns:
-        primary key
-        """
-        return inspect(self.__class__).primary_key[0].name
-
-    def construct_select_statement(
-        self,
-        compare_values: list,
-        select_keys: str | list[str] = None,
-        compare_keys: str | list[str] = None,
-        comparators: str | list = None,
-    ):
-        """
-        Function to construct a select statement
-        Args:
-            compare_values: Values to compare
-            select_keys: Keys to select. If None, defaults to primary key
-            compare_keys: Keys to compare. If None, defaults to primary key
-            comparators: Comparators : '__eq__', '__gt__', '__lt__', etc.
-            If not specified, defaults to '__eq__
-        Returns:
-        """
-
-        if compare_keys is None:
-            compare_keys = [self.get_primary_key()]
-        else:
-            if not isinstance(compare_keys, list):
-                compare_keys = [compare_keys]
-            if not isinstance(compare_values, list):
-                compare_values = [compare_values]
-            assert len(compare_keys) == len(compare_values)
-            columns = inspect(self.__class__).mapper.column_attrs
-            column_names = [c.key for c in columns]
-            for key in compare_keys:
-                if key not in column_names:
-                    err = f"{self.__tablename__} has no column {key}"
-                    raise ValueError(err)
-
-        if comparators is None:
-            comparators = ["__eq__"] * len(compare_keys)
-
-        if not isinstance(comparators, list):
-            comparators = [comparators]
-
-        assert len(compare_keys) == len(comparators)
-
-        logger.debug(
-            f"Checking if {compare_keys} {comparators} {compare_values} "
-            f"exists in {self.__tablename__}"
-        )
-
-        if select_keys is None:
-            select_items = [self.__class__]
-        else:
-            if not isinstance(select_keys, list):
-                select_keys = [select_keys]
-            select_items = [getattr(self.__class__, k) for k in select_keys]
-        stmt = select(*select_items).where(
-            and_(
-                *[
-                    getattr(self.__class__, key).__getattribute__(comparator)(value)
-                    for key, comparator, value in zip(
-                        compare_keys, comparators, compare_values
-                    )
-                ]
-            )
-        )
-
-        return stmt
-
-    def select_query(
-        self,
-        compare_values: list,
-        select_keys: str | list[str] = None,
-        compare_keys: str | list[str] = None,
-        comparators: str | list[str] = None,
-    ):
-        """
-        Run a select query
-        Args:
-            compare_values: Values to compare
-            select_keys: Keys to select. If None, defaults to primary key
-            compare_keys: Keys to compare. If None, defaults to primary key
-            comparators: Comparators : '__eq__', '__gt__', '__lt__', etc.
-            If None, defaults to '__eq__
-
-        Returns:
-            result of select query
-        """
-        engine = get_engine(db_name=self.db_name)
-        stmt = self.construct_select_statement(
-            compare_values=compare_values,
-            select_keys=select_keys,
-            compare_keys=compare_keys,
-            comparators=comparators,
-        )
-        with engine.connect() as conn:
-            res = conn.execute(stmt)
-            conn.commit()
-        return res.fetchall()
-
-    def exists(self, values, keys: str | list = None) -> bool:
-        """
-        Function to query a table and see whether an entry with key==value exists.
-        If key is None, key will default to the table's primary key.
-        Args:
-            values: = Value to match
-            keys: str = column name
-
-        Returns:
-            boolean
-        """
-        if not isinstance(keys, list):
-            keys = [keys]
-
-        if not isinstance(values, list):
-            values = [values]
-        assert len(values) == len(keys)
-        engine = get_engine(db_name=self.db_name)
-
-        logger.debug(f"Checking if {keys}=={values} exists in {self.__tablename__}")
-        stmt = self.construct_select_statement(
-            compare_values=values, compare_keys=keys, comparators=["__eq__"] * len(keys)
-        )
-        return _exists(stmt, engine=engine)
-
-
-def _execute(stmt, engine):
-    """
-    Checks if the pydantic-ified data exists the corresponding sql database
-    :return: bool
-    """
-    with engine.connect() as conn:
-        with conn.begin():
-            res = conn.execute(stmt).fetchall()
-    return res
-
-
-def _exists(stmt: Select, engine) -> bool:
-    """
-    Checks if the pydantic-ified data exists the corresponding sql database
-    :return: bool
-    """
-    return len(_execute(stmt, engine)) > 0
-
-
-ra_field: float = Field(title="RA (degrees)", ge=0.0, le=360.0)
-dec_field: float = Field(title="Dec (degrees)", ge=-90.0, le=90.0)
-alt_field: float = Field(title="Alt (degrees)", ge=0.0, le=90.0)
-az_field: float = Field(title="Az (degrees)", ge=0.0, le=360.0)
-date_field: date = Field()
diff --git a/mirar/processors/sqldatabase/postgres.py b/mirar/processors/sqldatabase/postgres.py
deleted file mode 100644
index 8b3e6a635..000000000
--- a/mirar/processors/sqldatabase/postgres.py
+++ /dev/null
@@ -1,575 +0,0 @@
-"""
-Module containing postgres util functions
-"""
-# pylint: disable=not-context-manager
-import logging
-from pathlib import Path
-from typing import Optional, Type
-
-import psycopg
-from psycopg import errors
-from psycopg.rows import Row
-from pydantic import ValidationError
-from sqlalchemy import inspect
-from sqlalchemy.exc import IntegrityError
-
-from mirar.data import DataBlock
-from mirar.errors import ProcessorError
-from mirar.processors.database.constraints import DBQueryConstraints
-from mirar.processors.sqldatabase.base_model import BaseDB
-from mirar.processors.sqldatabase.postgres_utils import (
-    ADMIN_PASSWORD,
-    ADMIN_USER,
-    DB_PASSWORD,
-    DB_PASSWORD_KEY,
-    DB_USER,
-    DB_USER_KEY,
-    PG_ADMIN_PWD_KEY,
-    PG_ADMIN_USER_KEY,
-    POSTGRES_DUPLICATE_PROTOCOLS,
-)
-
-logger = logging.getLogger(__name__)
-
-
-class DataBaseError(ProcessorError):
-    """Error relating to postgres interactions"""
-
-
-class PostgresUser:
-    """
-    Basic Postgres user class for executing functions
-    """
-
-    user_env_varaiable = DB_USER_KEY
-    pass_env_variable = DB_PASSWORD_KEY
-
-    def __init__(self, db_user: str = DB_USER, db_password: str = DB_PASSWORD):
-        self.db_user = db_user
-        self.db_password = db_password
-
-    def validate_credentials(self):
-        """
-        Checks that user credentials exist
-        :return: None
-        """
-        if self.db_user is None:
-            err = (
-                f"'db_user' is set as None. Please pass a db_user as an argument, "
-                f"or set the environment variable '{self.user_env_varaiable}'."
-            )
-            logger.error(err)
-            raise DataBaseError(err)
-
-        if self.db_password is None:
-            err = (
-                f"'db_password' is set as None. Please pass a password as an argument, "
-                f"or set the environment variable '{self.pass_env_variable}'."
-            )
-            logger.error(err)
-            raise DataBaseError(err)
-
-        # TODO check user exists
-
-    def run_sql_command_from_file(self, file_path: str | Path, db_name: str):
-        """
-        Execute SQL command from file
-
-        :param file_path: File to execute
-        :param db_name: name of database
-        :return: False
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            with open(file_path, "r", encoding="utf8") as sql_file:
-                conn.execute(sql_file.read())
-
-            logger.info(f"Executed sql commands from file {file_path}")
-
-    def export_to_db(
-        self,
-        value_dict: dict | DataBlock,
-        db_table: Type[BaseDB],
-        duplicate_protocol: str = "fail",
-    ) -> tuple[list, list]:
-        """
-        Export a list of fields in value dict to a batabase table
-
-        :param value_dict: dictionary/DataBlock/other dictonary-like object to export
-        :param db_table: table of DB to export to
-        :param duplicate_protocol: protocol for handling duplicates,
-            in "fail"/"ignore"/"replace"
-        :return:
-        """
-
-        assert duplicate_protocol in POSTGRES_DUPLICATE_PROTOCOLS
-
-        column_names = [
-            x for x in db_table.__dict__["__annotations__"] if x != "sql_model"
-        ]
-
-        column_dict = {}
-        for column in column_names:
-            column_dict[column] = value_dict[column]
-
-        try:
-            new = db_table(**column_dict)
-        except ValidationError as err:
-            logger.error(err)
-            raise DataBaseError from err
-
-        db_name = new.sql_model.db_name
-        primary_key = inspect(db_table.sql_model).primary_key[0]
-
-        sequence_key_names, sequence_values = [], []
-        try:
-            sequence_key_names, sequence_values = new.insert_entry()
-
-        except IntegrityError as exc:
-            if not isinstance(exc.orig, errors.UniqueViolation):
-                raise exc
-
-            if duplicate_protocol == "fail":
-                err = (
-                    f"Duplicate error, entry with {column_dict} "
-                    f"already exists in {db_name}."
-                )
-                logger.error(err)
-                raise errors.UniqueViolation from exc
-
-            if duplicate_protocol == "ignore":
-                logger.debug(
-                    f"Found duplicate entry in {db_name} - "
-                    f"{str(exc)}."
-                    f"Ignoring, no new entry made."
-                )
-                primary_key_val = value_dict[primary_key.name]
-                sequence_keys = new.get_sequence_keys()
-                sequence_key_names = [k.name for k in sequence_keys]
-                sequence_values = []
-                if len(sequence_keys) > 0:
-                    ret = new.sql_model().select_query(
-                        compare_values=[primary_key_val],
-                        compare_keys=[primary_key.name],
-                        select_keys=sequence_key_names,
-                    )
-                    sequence_values = [x[0] for x in ret]
-
-            if duplicate_protocol == "replace":
-                logger.debug(f"Conflict at {exc.orig.diag.constraint_name}")
-                logger.debug(
-                    f"Found duplicate entry in {db_name} - "
-                    f"{str(exc)}."
-                    f"Replacing with a new entry."
-                )
-                primary_key_val = value_dict[primary_key.name]
-                sequence_key_names, sequence_values = new.update_entry(
-                    primary_key_val=primary_key_val
-                )
-
-        return sequence_key_names, sequence_values
-
-    def execute_query(self, sql_query: str, db_name: str) -> list[Row]:
-        """
-        Generically execute SQL query
-
-        :param sql_query: SQL query to execute
-        :param db_name: db name
-        :return: rows from db
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            logger.debug(f"Query: {sql_query}")
-
-            with conn.execute(sql_query) as cursor:
-                query_output = cursor.fetchall()
-
-        return query_output
-
-    def import_from_db(
-        self,
-        db_name: str,
-        db_table: str,
-        db_output_columns: str | list[str],
-        output_alias_map: Optional[str | list[str]] = None,
-        max_num_results: Optional[int] = None,
-        db_constraints: Optional[DBQueryConstraints] = None,
-    ) -> list[dict]:
-        """Query an SQL database with constraints, and return a list of dictionaries.
-        One dictionary per entry returned from the query.
-
-        #TODO check admin
-
-        Parameters
-        ----------
-        db_name: Name of database to query
-        db_table: Name of database table to query
-        db_output_columns: Name(s) of columns to return for matched database entries
-        output_alias_map: Alias to assign for each output column
-        max_num_results: Maximum number of results to return
-
-        Returns
-        -------
-        A list of dictionaries (one per entry)
-        """
-
-        if not isinstance(db_output_columns, list):
-            db_output_columns = [db_output_columns]
-
-        if output_alias_map is None:
-            output_alias_map = db_output_columns
-
-        if not isinstance(output_alias_map, list):
-            output_alias_map = [output_alias_map]
-
-        assert len(output_alias_map) == len(db_output_columns)
-
-        all_query_res = []
-
-        if db_constraints is not None:
-            constraints = db_constraints.parse_constraints()
-        else:
-            constraints = ""
-
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            sql_query = f"""
-            SELECT {', '.join(db_output_columns)} from {db_table}
-                WHERE {constraints}
-            """
-
-            if max_num_results is not None:
-                sql_query += f" LIMIT {max_num_results}"
-
-            sql_query += ";"
-
-            logger.debug(f"Query: {sql_query}")
-
-            with conn.execute(sql_query) as cursor:
-                query_output = cursor.fetchall()
-
-            for entry in query_output:
-                assert len(entry) == len(db_output_columns)
-
-                query_res = {}
-
-                for i, key in enumerate(output_alias_map):
-                    query_res[key] = entry[i]
-
-                all_query_res.append(query_res)
-
-        return all_query_res
-
-    def crossmatch_with_database(
-        self,
-        db_name: str,
-        db_table: str,
-        db_output_columns: str | list[str],
-        ra: float,
-        dec: float,
-        crossmatch_radius_arcsec: float,
-        output_alias_map: Optional[dict] = None,
-        ra_field_name: str = "ra",
-        dec_field_name: str = "dec",
-        query_distance_bool: bool = False,
-        q3c_bool: bool = False,
-        query_constraints: Optional[DBQueryConstraints] = None,
-        order_field_name: Optional[str] = None,
-        num_limit: Optional[int] = None,
-    ) -> list[dict]:
-        """
-        Crossmatch a given spatial position (ra/dec) with sources in a database,
-        and returns a list of matches
-
-        #TODO: check admin
-
-        :param db_name: name of db to query
-        :param db_table: name of db table
-        :param db_output_columns: columns to return
-        :param output_alias_map: mapping for renaming columns
-        :param ra: RA
-        :param dec: dec
-        :param crossmatch_radius_arcsec: radius for crossmatch
-        :param ra_field_name: name of ra column in database
-        :param dec_field_name: name of dec column in database
-        :param query_distance_bool: boolean where to return crossmatch distance
-        :param q3c_bool: boolean whether to use q3c_bool
-        :param order_field_name: field to order result by
-        :param num_limit: limit on sql query
-        :return: list of query result dictionaries
-        """
-
-        if output_alias_map is None:
-            output_alias_map = {}
-            for col in db_output_columns:
-                output_alias_map[col] = col
-
-        crossmatch_radius_deg = crossmatch_radius_arcsec / 3600.0
-
-        if q3c_bool:
-            constraints = (
-                f"q3c_radial_query({ra_field_name},{dec_field_name},"
-                f"{ra},{dec},{crossmatch_radius_deg}) "
-            )
-        else:
-            ra_min = ra - crossmatch_radius_deg
-            ra_max = ra + crossmatch_radius_deg
-            dec_min = dec - crossmatch_radius_deg
-            dec_max = dec + crossmatch_radius_deg
-            constraints = (
-                f" {ra_field_name} between {ra_min} and {ra_max} AND "
-                f"{dec_field_name} between {dec_min} and {dec_max} "
-            )
-
-        if query_constraints is not None:
-            constraints += f"""AND {query_constraints.parse_constraints()}"""
-
-        select = f""" {'"' + '","'.join(db_output_columns) + '"'}"""
-        if query_distance_bool:
-            if q3c_bool:
-                select = (
-                    f"q3c_dist({ra_field_name},{dec_field_name},{ra},{dec}) AS xdist,"
-                    + select
-                )
-            else:
-                select = f"""{ra_field_name} - ra AS xdist,""" + select
-
-        query = f"""SELECT {select} FROM {db_table} WHERE {constraints}"""
-
-        if order_field_name is not None:
-            query += f""" ORDER BY {order_field_name}"""
-        if num_limit is not None:
-            query += f""" LIMIT {num_limit}"""
-
-        query += ";"
-
-        query_output = self.execute_query(query, db_name)
-        all_query_res = []
-
-        for entry in query_output:
-            if not query_distance_bool:
-                assert len(entry) == len(db_output_columns)
-            else:
-                assert len(entry) == len(db_output_columns) + 1
-            query_res = {}
-            for i, key in enumerate(output_alias_map):
-                query_res[key] = entry[i]
-                if query_distance_bool:
-                    query_res["dist"] = entry["xdist"]
-            all_query_res.append(query_res)
-        return all_query_res
-
-    def check_if_exists(
-        self, check_command: str, check_value: str, db_name: str = "postgres"
-    ) -> bool:
-        """
-        Check if a user account exists
-
-        :param check_command if a user/database/table exists
-        :param check_value: username to check
-        :param db_name: name of database to query
-        :return: boolean
-        """
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            data = conn.execute(check_command).fetchall()
-        existing_user_names = [x[0] for x in data]
-        logger.debug(f"Found the following values: {existing_user_names}")
-
-        return check_value in existing_user_names
-
-    def check_if_db_exists(self, db_name: str) -> bool:
-        """
-        Check if a user account exists
-
-        :param db_name: database to check
-        :return: boolean
-        """
-
-        check_command = """SELECT datname FROM pg_database;"""
-
-        db_exist_bool = self.check_if_exists(
-            check_command=check_command,
-            check_value=db_name,
-            db_name="postgres",
-        )
-
-        logger.debug(f"Database '{db_name}' does {['not ', ''][db_exist_bool]} exist")
-
-        return db_exist_bool
-
-    def create_db(self, db_name: str):
-        """
-        Creates a database using credentials
-
-        :param db_name: DB to create
-        :return: None
-        """
-
-        with psycopg.connect(
-            f"dbname=postgres user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            sql = f"""CREATE database {db_name}"""
-            conn.execute(sql)
-            logger.info(f"Created db {db_name}")
-
-    def check_if_table_exists(self, db_name: str, db_table: str) -> bool:
-        """
-        Check if a db table account exists
-
-        :param db_name: database to check
-        :param db_table: table to check
-        :return: boolean
-        """
-
-        check_command = (
-            "SELECT table_name FROM information_schema.tables "
-            "WHERE table_schema='public';"
-        )
-
-        table_exist_bool = self.check_if_exists(
-            check_command=check_command,
-            check_value=db_table,
-            db_name=db_name,
-        )
-
-        logger.debug(f"Table '{db_table}' does {['not ', ''][table_exist_bool]} exist")
-
-        return table_exist_bool
-
-    def modify_db_entry(
-        self,
-        db_name: str,
-        db_table: str,
-        db_constraints: DBQueryConstraints,
-        value_dict: dict | DataBlock,
-        db_alter_columns: str | list[str],
-        return_columns: Optional[str | list[str]] = None,
-    ) -> list[Row]:
-        """
-        Modify a db entry
-
-        :param db_name: name of db
-        :param db_table: Name of table
-        :param value_dict: dict-like object to provide updated values
-        :param db_alter_columns: columns to alter in db
-        :param return_columns: columns to return
-        :return: db query (return columns)
-        """
-
-        if not isinstance(db_alter_columns, list):
-            db_alter_columns = [db_alter_columns]
-
-        if return_columns is None:
-            return_columns = db_alter_columns
-        if not isinstance(return_columns, list):
-            return_columns = [return_columns]
-
-        constraints = db_constraints.parse_constraints()
-
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-
-            db_alter_values = [str(value_dict[c]) for c in db_alter_columns]
-
-            alter_values_txt = [
-                f"{db_alter_columns[ind]}='{db_alter_values[ind]}'"
-                for ind in range(len(db_alter_columns))
-            ]
-
-            sql_query = (
-                f"UPDATE {db_table} SET {', '.join(alter_values_txt)} "
-                f"WHERE {constraints}"
-            )
-
-            if len(return_columns) > 0:
-                logger.debug(return_columns)
-                sql_query += f""" RETURNING {', '.join(return_columns)}"""
-            sql_query += ";"
-            query_output = self.execute_query(sql_query, db_name)
-
-        return query_output
-
-
-class PostgresAdmin(PostgresUser):
-    """
-    An Admin postgres user, with additional functionality for creatying new users
-    """
-
-    user_env_varaiable = PG_ADMIN_USER_KEY
-    pass_env_variable = PG_ADMIN_PWD_KEY
-
-    def __init__(self, db_user: str = ADMIN_USER, db_password: str = ADMIN_PASSWORD):
-        super().__init__(db_user=db_user, db_password=db_password)
-
-    def create_new_user(self, new_db_user: str, new_password: str):
-        """
-        Create a new postgres user
-
-        :param new_db_user: new username
-        :param new_password: new user password
-        :return: None
-        """
-
-        with psycopg.connect(
-            f"dbname=postgres user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            command = f"CREATE ROLE {new_db_user} WITH password '{new_password}' LOGIN;"
-            conn.execute(command)
-
-    def grant_privileges(self, db_name: str, db_user: str, schema: str = "public"):
-        """
-        Grant privilege to user on database
-
-        :param db_name: name of database
-        :param db_user: username to grant privileges for db_user
-        :return: None
-        """
-        with psycopg.connect(
-            f"dbname=postgres user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            command = f"""GRANT ALL PRIVILEGES ON DATABASE {db_name} TO {db_user};"""
-            conn.execute(command)
-            command = f"""ALTER DATABASE {db_name} OWNER TO {db_user}"""
-            conn.execute(command)
-
-        with psycopg.connect(
-            f"dbname={db_name} user={self.db_user} password={self.db_password}"
-        ) as conn:
-            conn.autocommit = True
-            command = f"""GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA {schema}
-            TO {db_user};"""
-            conn.execute(command)
-            command = f"""GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA {schema}
-                        TO {db_user};"""
-            conn.execute(command)
-
-    def check_if_user_exists(self, user_name: str) -> bool:
-        """
-        Check if a user account exists
-
-        :param user_name: username to check
-        :return: boolean
-        """
-        check_command = """SELECT usename FROM pg_user;"""
-
-        user_exist_bool = self.check_if_exists(
-            check_command=check_command,
-            check_value=user_name,
-        )
-
-        logger.debug(f"User '{user_name}' does {['not ', ''][user_exist_bool]} exist")
-
-        return user_exist_bool
diff --git a/mirar/processors/sqldatabase/postgres_utils.py b/mirar/processors/sqldatabase/postgres_utils.py
deleted file mode 100644
index ca8629430..000000000
--- a/mirar/processors/sqldatabase/postgres_utils.py
+++ /dev/null
@@ -1,57 +0,0 @@
-"""
-Module for postgres utilities
-"""
-import logging
-import os
-
-import numpy as np
-import psycopg
-
-logger = logging.getLogger(__name__)
-
-
-DB_USER_KEY = "DB_USER"
-DB_PASSWORD_KEY = "DB_PWD"
-
-PG_ADMIN_USER_KEY = "PG_ADMIN_USER"
-PG_ADMIN_PWD_KEY = "PG_ADMIN_PWD"
-
-DB_USER = os.getenv(DB_USER_KEY)
-DB_PASSWORD = os.getenv(DB_PASSWORD_KEY)
-
-ADMIN_USER = os.getenv(PG_ADMIN_USER_KEY, DB_USER)
-ADMIN_PASSWORD = os.getenv(PG_ADMIN_PWD_KEY, DB_PASSWORD)
-
-POSTGRES_DUPLICATE_PROTOCOLS = ["fail", "ignore", "replace"]
-
-
-def get_sequence_key_names_from_table(
-    db_table: str,
-    db_name: str,
-    db_user: str = DB_USER,
-    db_password: str = DB_PASSWORD,
-) -> np.ndarray:
-    """
-    Gets sequence keys of db table
-
-    :param db_table: database table to use
-    :param db_name: database name
-    :param db_user: database user
-    :param db_password: dataname password
-    :return: numpy array of keys
-
-    """
-    with psycopg.connect(
-        f"dbname={db_name} user={db_user} password={db_password}"
-    ) as conn:
-        conn.autocommit = True
-        sequences = [
-            x[0]
-            for x in conn.execute(
-                "SELECT c.relname FROM pg_class c WHERE c.relkind = 'S';"
-            ).fetchall()
-        ]
-        seq_tables = np.array([x.split("_")[0] for x in sequences])
-        seq_columns = np.array([x.split("_")[1] for x in sequences])
-        table_sequence_keys = seq_columns[(seq_tables == db_table)]
-    return table_sequence_keys
diff --git a/mirar/processors/xmatch.py b/mirar/processors/xmatch.py
index d82b300bf..3a5986e17 100644
--- a/mirar/processors/xmatch.py
+++ b/mirar/processors/xmatch.py
@@ -53,7 +53,7 @@ def _apply_to_sources(
                 for num in range(self.catalog.num_sources):
                     colname = catalog.column_names[key]
                     candidate_table[colname + f"{num + 1}"] = np.array(
-                        np.zeros(len(candidate_table)) - 99,
+                        np.nan,
                         dtype=catalog.column_dtypes[colname],
                     )
             nmatch_colname = f"nmtch{self.catalog.abbreviation}"
@@ -67,6 +67,8 @@ def _apply_to_sources(
 
                 candidate_table.at[query_ind, nmatch_colname] = len(results)
 
+            candidate_table = candidate_table.replace({np.nan: None})
+
             source_list.set_data(candidate_table)
 
         return batch
diff --git a/mirar/references/base_reference_generator.py b/mirar/references/base_reference_generator.py
index 22aab5cf7..2d2257002 100644
--- a/mirar/references/base_reference_generator.py
+++ b/mirar/references/base_reference_generator.py
@@ -12,6 +12,7 @@
 
 from mirar.data import Image, ImageBatch
 from mirar.data.utils import get_corners_ra_dec_from_header, get_image_center_wcs_coords
+from mirar.database.base_model import BaseDB
 from mirar.errors import ProcessorError
 from mirar.io import (
     MissingCoreFieldError,
@@ -33,9 +34,8 @@
 )
 from mirar.processors.astromatic.sextractor.sextractor import Sextractor
 from mirar.processors.astromatic.swarp import Swarp
+from mirar.processors.database.database_inserter import DatabaseImageInserter
 from mirar.processors.photcal import PhotCalibrator
-from mirar.processors.sqldatabase.base_model import BaseDB
-from mirar.processors.sqldatabase.database_exporter import DatabaseImageExporter
 
 logger = logging.getLogger(__name__)
 
@@ -175,10 +175,9 @@ def get_reference_image(self, image: Image) -> Image:
             save_hdu_as_fits(ref_hdu, output_path)
 
         if self.write_to_db:
-            dbexporter = DatabaseImageExporter(
+            dbexporter = DatabaseImageInserter(
                 db_table=self.write_db_table,
                 duplicate_protocol=self.duplicate_protocol,
-                q3c_bool=self.q3c_bool,
             )
             ref_image_batch = ImageBatch([ref_image])
             _ = dbexporter.apply(ref_image_batch)
diff --git a/mirar/references/local.py b/mirar/references/local.py
index b2354dd40..13d2441ac 100644
--- a/mirar/references/local.py
+++ b/mirar/references/local.py
@@ -8,8 +8,8 @@
 from astropy.io import fits
 
 from mirar.data import Image
+from mirar.database.base_model import BaseDB
 from mirar.paths import LATEST_WEIGHT_SAVE_KEY
-from mirar.processors.sqldatabase.base_model import BaseDB
 from mirar.references.base_reference_generator import BaseReferenceGenerator
 
 logger = logging.getLogger(__name__)
diff --git a/mirar/references/wfcam/wfcam_query.py b/mirar/references/wfcam/wfcam_query.py
index dc46a8a1d..41c52fc8b 100644
--- a/mirar/references/wfcam/wfcam_query.py
+++ b/mirar/references/wfcam/wfcam_query.py
@@ -19,11 +19,13 @@
 
 from mirar.data import Image, ImageBatch
 from mirar.data.utils import get_image_center_wcs_coords
+from mirar.database.base_model import BaseDB
+from mirar.database.constraints import DBQueryConstraints
+from mirar.database.transactions import select_from_table
 from mirar.errors import ProcessorError
 from mirar.io import open_raw_image, save_fits
 from mirar.paths import BASE_NAME_KEY, LATEST_SAVE_KEY, get_output_dir, get_output_path
-from mirar.processors.sqldatabase.base_model import BaseDB
-from mirar.processors.sqldatabase.database_exporter import DatabaseImageExporter
+from mirar.processors.database import DatabaseImageInserter
 from mirar.references.wfcam.utils import (
     COMPID_KEY,
     EXTENSION_ID_KEY,
@@ -194,6 +196,7 @@ def __init__(
         self.query_db_table = query_db_table
         self.use_db_for_component_queries = use_db_for_component_queries
         self.skip_online_query = skip_online_query
+        self.dbexporter = DatabaseImageInserter(db_table=self.query_db_table)
 
         if self.use_db_for_component_queries:
             if self.components_db_table is None:
@@ -360,8 +363,7 @@ def run_wfau_query(
                     for img in downloaded_images:
                         img[QUERY_RA_KEY] = crd.ra.deg
                         img[QUERY_DEC_KEY] = crd.dec.deg
-                    dbexporter = DatabaseImageExporter(db_table=self.query_db_table)
-                    _ = dbexporter.apply(ImageBatch(downloaded_images))
+                    self.dbexporter.apply(ImageBatch(downloaded_images))
 
                 qexists_list = [query_exists] * len(imagepaths)
                 ra_list, dec_list = [crd.ra.deg] * len(imagepaths), [crd.dec.deg] * len(
@@ -412,7 +414,6 @@ def download_wfcam_archive_images(
     use_local_database: bool = False,
     components_table: Type[BaseDB] = None,
     duplicate_protocol: str = "ignore",
-    q3c_bool: bool = False,
 ) -> list[Path]:
     """
     Download the image from UKIRT server. Optionally, check if the image exists locally
@@ -497,10 +498,9 @@ def download_wfcam_archive_images(
             wfcam_image[LATEST_SAVE_KEY] = imagepath.as_posix()
 
             if use_local_database:
-                dbexporter = DatabaseImageExporter(
+                dbexporter = DatabaseImageInserter(
                     db_table=components_table,
                     duplicate_protocol=duplicate_protocol,
-                    q3c_bool=q3c_bool,
                 )
                 wfcam_db_batch = dbexporter.apply(ImageBatch([wfcam_image]))
                 wfcam_image = wfcam_db_batch[0]
@@ -532,28 +532,31 @@ def check_query_exists_locally(
     Returns:
         :return: list of savepaths
     """
-    results = query_table.sql_model().select_query(
-        select_keys=COMPID_KEY.lower(),
-        compare_values=[query_ra, query_dec, query_filt],
-        compare_keys=[
-            QUERY_RA_KEY.lower(),
-            QUERY_DEC_KEY.lower(),
-            QUERY_FILT_KEY.lower(),
-        ],
-        comparators=["__eq__", "__eq__", "__eq__"],
+
+    constraints = DBQueryConstraints(
+        columns=[QUERY_RA_KEY, QUERY_DEC_KEY, QUERY_FILT_KEY],
+        accepted_values=[query_ra, query_dec, query_filt],
+    )
+    results = select_from_table(
+        db_constraints=constraints,
+        sql_table=query_table.sql_model,
+        output_columns=[COMPID_KEY],
     )
+
     logger.debug(results)
     savepaths = []
-    # TODO - run a joint query to get the savepaths
     if len(results) > 0:
         savepaths = []
         compids = [x[0] for x in results]
         for compid in compids:
-            comp_results = components_table.sql_model().select_query(
-                select_keys="savepath",
-                compare_values=[compid],
-                compare_keys=[COMPID_KEY.lower()],
-                comparators=["__eq__"],
+            constraints = DBQueryConstraints(
+                columns=[COMPID_KEY],
+                accepted_values=[compid],
+            )
+            comp_results = select_from_table(
+                db_constraints=constraints,
+                sql_table=components_table.sql_model,
+                output_columns=["savepath"],
             )
             if len(comp_results) == 0:
                 raise WFAUQueryDBError(
@@ -561,7 +564,7 @@ def check_query_exists_locally(
                     "a query corresponding to it exists. The query"
                     "table is likely out of sync with the component"
                 )
-            savepaths.append(Path(comp_results[0][0]))
+            savepaths.append(Path(comp_results["savepath"].iloc[0].value))
     return savepaths
 
 
@@ -579,16 +582,23 @@ def get_locally_existing_overlap_images(
     Returns:
         :return: list of savepaths
     """
-    savepaths = []
-    results = components_table.sql_model().select_query(
-        select_keys=["savepath"],
-        compare_values=[query_ra, query_ra, query_dec, query_dec, query_filt],
-        compare_keys=["ra0_0", "ra1_1", "dec0_0", "dec1_1", "filter"],
-        comparators=["__le__", "__ge__", "__ge__", "__le__", "__eq__"],
+
+    constraints = DBQueryConstraints(
+        columns=["ra0_0", "ra1_1", "dec0_0", "dec1_1", "filter"],
+        accepted_values=[query_ra, query_ra, query_dec, query_dec, query_filt],
+        comparison_types=["<=", ">=", ">=", "<=", "="],
+    )
+
+    results = select_from_table(
+        db_constraints=constraints,
+        sql_table=components_table.sql_model,
+        output_columns=["savepath"],
     )
+
     logger.debug(results)
+    savepaths = []
     if len(results) > 0:
-        savepaths = [Path(x[0]) for x in results]
+        savepaths = [Path(x) for x in results["savepath"].tolist()]
     return savepaths
 
 
@@ -607,23 +617,23 @@ def check_multiframe_exists_locally(
     Returns:
         :return: list of savepaths
     """
-    results = db_table.sql_model().select_query(
-        select_keys=["savepath"],
-        compare_values=[
-            multiframe_id,
-            extension_id,
-        ],
-        compare_keys=[
-            MULTIFRAME_ID_KEY.lower(),
-            EXTENSION_ID_KEY.lower(),
-        ],
-        comparators=["__eq__", "__eq__"],
+
+    db_constraints = DBQueryConstraints(
+        columns=[MULTIFRAME_ID_KEY.lower(), EXTENSION_ID_KEY.lower()],
+        accepted_values=[multiframe_id, extension_id],
+    )
+
+    results = select_from_table(
+        db_constraints=db_constraints,
+        sql_table=db_table.sql_model,
+        output_columns=["savepath"],
     )
+
     logger.debug(results)
     if len(results) == 0:
         savepaths = []
     else:
-        savepaths = [Path(x[0]) for x in results]
+        savepaths = [Path(x) for x in results["savepath"].tolist()]
     return savepaths
 
 
diff --git a/mirar/references/wfcam/wfcam_stack.py b/mirar/references/wfcam/wfcam_stack.py
index ec1e08951..654b27d7e 100644
--- a/mirar/references/wfcam/wfcam_stack.py
+++ b/mirar/references/wfcam/wfcam_stack.py
@@ -20,12 +20,12 @@
 import numpy as np
 
 from mirar.data import Image, ImageBatch
+from mirar.database.base_model import BaseDB
 from mirar.paths import BASE_NAME_KEY, ZP_KEY, ZP_STD_KEY
 from mirar.processors.astromatic.sextractor.sextractor import Sextractor
 from mirar.processors.astromatic.swarp import Swarp
 from mirar.processors.base_processor import ImageHandler
 from mirar.processors.photcal import PhotCalibrator
-from mirar.processors.sqldatabase.base_model import BaseDB
 from mirar.references.base_reference_generator import BaseStackReferenceGenerator
 from mirar.references.wfcam.wfcam_query import BaseWFCAMQuery
 
diff --git a/mirar/utils/sql.py b/mirar/utils/sql.py
deleted file mode 100644
index d16f73e0b..000000000
--- a/mirar/utils/sql.py
+++ /dev/null
@@ -1,58 +0,0 @@
-"""
-Util functions for database interactions
-"""
-from sqlalchemy import DDL, Engine, NullPool, create_engine
-
-from mirar.processors.database.postgres import DB_PASSWORD, DB_USER
-
-
-def get_engine(
-    db_user: str = DB_USER,
-    db_password: str = DB_PASSWORD,
-    db_name: str = "summer",
-) -> Engine:
-    """
-    Function to create a postgres engine
-
-    :param db_user: User for db
-    :param db_password: password for db
-    :param db_name: name of db
-    :return: sqlalchemy engine
-    """
-
-    return create_engine(
-        f"postgresql+psycopg://{db_user}:{db_password}" f"@/{db_name}",
-        future=True,
-        poolclass=NullPool,
-    )
-
-
-def create_q3c_extension(
-    table_name: str, ra_column_name: str, dec_column_name: str, conn=None
-):
-    """
-    Function to create q3c extension and index on table
-
-    :param table_name: Name of table
-    :param ra_column_name: ra column name
-    :param dec_column_name: dec column name
-    :param conn: connection to db
-    :return:
-    """
-
-    trig_ddl = DDL(
-        "CREATE EXTENSION IF NOT EXISTS q3c;"
-        f"CREATE INDEX ON {table_name} "
-        f"(q3c_ang2ipix({ra_column_name}, {dec_column_name}));"
-        f"CLUSTER {table_name} USING {table_name}_q3c_ang2ipix_idx;"
-        f"ANALYZE {table_name};"
-    )
-
-    if conn is None:
-        engine = get_engine()
-        with engine.connect() as new_conn:
-            new_conn.execute(trig_ddl)
-            new_conn.commit()
-    else:
-        conn.execute(trig_ddl)
-        conn.commit()
diff --git a/poetry.lock b/poetry.lock
index e3df4bfdd..c77d0974e 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -2292,7 +2292,6 @@ optional = false
 python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*, !=3.5.*, !=3.6.*"
 files = [
     {file = "jsonpointer-2.4-py2.py3-none-any.whl", hash = "sha256:15d51bba20eea3165644553647711d150376234112651b4f1811022aecad7d7a"},
-    {file = "jsonpointer-2.4.tar.gz", hash = "sha256:585cee82b70211fa9e6043b7bb89db6e1aa49524340dde8ad6b63206ea689d88"},
 ]
 
 [[package]]
@@ -5145,6 +5144,34 @@ postgresql-psycopgbinary = ["psycopg[binary] (>=3.0.7)"]
 pymysql = ["pymysql"]
 sqlcipher = ["sqlcipher3-binary"]
 
+[[package]]
+name = "sqlalchemy-utils"
+version = "0.41.1"
+description = "Various utility functions for SQLAlchemy."
+optional = false
+python-versions = ">=3.6"
+files = [
+    {file = "SQLAlchemy-Utils-0.41.1.tar.gz", hash = "sha256:a2181bff01eeb84479e38571d2c0718eb52042f9afd8c194d0d02877e84b7d74"},
+    {file = "SQLAlchemy_Utils-0.41.1-py3-none-any.whl", hash = "sha256:6c96b0768ea3f15c0dc56b363d386138c562752b84f647fb8d31a2223aaab801"},
+]
+
+[package.dependencies]
+SQLAlchemy = ">=1.3"
+
+[package.extras]
+arrow = ["arrow (>=0.3.4)"]
+babel = ["Babel (>=1.3)"]
+color = ["colour (>=0.0.4)"]
+encrypted = ["cryptography (>=0.6)"]
+intervals = ["intervals (>=0.7.1)"]
+password = ["passlib (>=1.6,<2.0)"]
+pendulum = ["pendulum (>=2.0.5)"]
+phone = ["phonenumbers (>=5.9.2)"]
+test = ["Jinja2 (>=2.3)", "Pygments (>=1.2)", "backports.zoneinfo", "docutils (>=0.10)", "flake8 (>=2.4.0)", "flexmock (>=0.9.7)", "isort (>=4.2.2)", "pg8000 (>=1.12.4)", "psycopg (>=3.1.8)", "psycopg2 (>=2.5.1)", "psycopg2cffi (>=2.8.1)", "pymysql", "pyodbc", "pytest (>=2.7.1)", "python-dateutil (>=2.6)", "pytz (>=2014.2)"]
+test-all = ["Babel (>=1.3)", "Jinja2 (>=2.3)", "Pygments (>=1.2)", "arrow (>=0.3.4)", "backports.zoneinfo", "colour (>=0.0.4)", "cryptography (>=0.6)", "docutils (>=0.10)", "flake8 (>=2.4.0)", "flexmock (>=0.9.7)", "furl (>=0.4.1)", "intervals (>=0.7.1)", "isort (>=4.2.2)", "passlib (>=1.6,<2.0)", "pendulum (>=2.0.5)", "pg8000 (>=1.12.4)", "phonenumbers (>=5.9.2)", "psycopg (>=3.1.8)", "psycopg2 (>=2.5.1)", "psycopg2cffi (>=2.8.1)", "pymysql", "pyodbc", "pytest (>=2.7.1)", "python-dateutil", "python-dateutil (>=2.6)", "pytz (>=2014.2)"]
+timezone = ["python-dateutil"]
+url = ["furl (>=0.4.1)"]
+
 [[package]]
 name = "stack-data"
 version = "0.6.2"
@@ -5751,4 +5778,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 [metadata]
 lock-version = "2.0"
 python-versions = ">=3.10.0,<3.12"
-content-hash = "9269c985725dbca5be5e5ec61799fd120e8098fd4a26796c906b0fdde1904fcf"
+content-hash = "511287510d0642911accce4c62e1b357eeb05fc42cfe9233cdf46b5919c7d3a9"
diff --git a/pyproject.toml b/pyproject.toml
index 0dce7b96c..cc583248b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -49,6 +49,7 @@ pylint-pydantic = ">=0.1.6,<0.3.0"
 astrosurveyutils = "^0.0.1"
 wintertoo = "^0.3.9"
 urllib3 = "^1.26.16"
+sqlalchemy-utils = "^0.41.1"
 
 [tool.poetry.group.docs.dependencies]
 sphinx = "^7.0.1"


Test patch:
diff --git a/tests/test_forced_photometry.py b/tests/test_forced_photometry.py
index 1fe02fef6..ae3080dab 100644
--- a/tests/test_forced_photometry.py
+++ b/tests/test_forced_photometry.py
@@ -33,7 +33,7 @@
 EXPECTED_DATAFRAME_VALUES = {
     "magpsf": [16.957492],
     "magap": [17.014131],
-    "objectId": ["ZTF21aagppzg"],
+    "objectid": ["ZTF21aagppzg"],
 }
 test_fp_configuration = (
     [