PKGBUILD/pkg/repository/postgres.go

1103 lines
26 KiB
Go

package repository
import (
"context"
"database/sql"
"errors"
"fmt"
"net/url"
"time"
"git.cryptic.systems/volker.raschek/flucky/pkg/repository/postgres"
"git.cryptic.systems/volker.raschek/flucky/pkg/types"
"git.cryptic.systems/volker.raschek/go-logger"
"github.com/golang-migrate/migrate/v4"
"github.com/johejo/golang-migrate-extra/source/iofs"
)
type PostgresOpts struct {
DatabaseURL *url.URL
Logger logger.Logger
}
func (o *PostgresOpts) Validate() error {
for k, v := range map[string]interface{}{
"DatabaseURL": o.DatabaseURL,
"Logger": o.Logger,
} {
if v == nil {
return fmt.Errorf("%v does not be empty or nil", k)
}
}
if o.DatabaseURL.Scheme != "postgres" {
return fmt.Errorf("Unsupported database scheme %v for database backend postgres", o.DatabaseURL.Scheme)
}
return nil
}
func NewPostgres(opts PostgresOpts) (Repository, error) {
if err := opts.Validate(); err != nil {
return nil, err
}
sqlDBO, err := sql.Open(opts.DatabaseURL.Scheme, opts.DatabaseURL.String())
if err != nil {
return nil, err
}
return &Postgres{
databaseURL: opts.DatabaseURL,
dbo: sqlDBO,
flogger: opts.Logger,
}, nil
}
// Postgres implementation
type Postgres struct {
databaseURL *url.URL
dbo *sql.DB
flogger logger.Logger
}
// AddDevices into the database
func (d *Postgres) AddDevices(ctx context.Context, devices ...*types.Device) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
err = d.insertDevices(tx, devices...)
if err != nil {
return err
}
return tx.Commit()
}
func (d *Postgres) insertDevices(tx *sql.Tx, devices ...*types.Device) error {
stmt, err := tx.Prepare(postgres.InsertDeviceSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, device := range devices {
if device.CreationDate.Equal(time.Time{}) {
device.CreationDate = time.Now()
}
_, err = stmt.Exec(
&device.ID,
&device.Name,
&device.Location,
&device.CreationDate,
&device.UpdateDate,
)
if err != nil {
return fmt.Errorf("Failed to execute statement: %v", err)
}
}
return nil
}
func (d *Postgres) AddOrUpdateDevices(ctx context.Context, devices ...*types.Device) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
err = d.insertOrUpdateDevices(tx, devices...)
if err != nil {
return err
}
return tx.Commit()
}
func (d *Postgres) insertOrUpdateDevices(tx *sql.Tx, devices ...*types.Device) error {
stmt, err := tx.Prepare(postgres.InsertOrUpdateDeviceSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %w", err)
}
defer func() { _ = stmt.Close() }()
for _, device := range devices {
if device.CreationDate.Equal(time.Time{}) {
device.CreationDate = time.Now()
}
_, err = stmt.Exec(
&device.ID,
&device.Name,
&device.Location,
&device.CreationDate,
&device.UpdateDate,
)
if err != nil {
return fmt.Errorf("Failed to execute statement: %v", err)
}
}
return nil
}
// AddMeasuredValues into the database
func (d *Postgres) AddMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error {
splittedMeasuredValues := make(map[types.MeasuredValueType][]*types.MeasuredValue, 0)
for _, measuredValue := range measuredValues {
if _, ok := splittedMeasuredValues[measuredValue.ValueType]; !ok {
splittedMeasuredValues[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
}
splittedMeasuredValues[measuredValue.ValueType] = append(splittedMeasuredValues[measuredValue.ValueType], measuredValue)
}
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
for measuredValueType, measuredValues := range splittedMeasuredValues {
var queryFile string
switch measuredValueType {
case types.Humidity:
queryFile = postgres.InsertHumiditySQL
case types.Pressure:
queryFile = postgres.InsertPressureSQL
case types.Temperature:
queryFile = postgres.InsertTemperatureSQL
default:
return fmt.Errorf("Measured value type %v not supported", measuredValueType)
}
err := d.insertMeasuredValues(tx, queryFile, measuredValues...)
if err != nil {
return err
}
}
return tx.Commit()
}
func (d *Postgres) insertMeasuredValues(tx *sql.Tx, query string, measuredValues ...*types.MeasuredValue) error {
stmt, err := tx.Prepare(query)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, measuredValue := range measuredValues {
if measuredValue.CreationDate.Equal(time.Time{}) {
measuredValue.CreationDate = time.Now()
}
_, err := stmt.Exec(
&measuredValue.ID,
&measuredValue.Value,
&measuredValue.Date,
&measuredValue.SensorID,
&measuredValue.CreationDate,
&measuredValue.UpdateDate,
)
if err != nil {
return fmt.Errorf("Failed to execute statement: %v", err)
}
}
return nil
}
// AddOrUpdateMeasuredValues into the database
func (d *Postgres) AddOrUpdateMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error {
splittedMeasuredValues := make(map[types.MeasuredValueType][]*types.MeasuredValue, 0)
for _, measuredValue := range measuredValues {
if _, ok := splittedMeasuredValues[measuredValue.ValueType]; !ok {
splittedMeasuredValues[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
}
splittedMeasuredValues[measuredValue.ValueType] = append(splittedMeasuredValues[measuredValue.ValueType], measuredValue)
}
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
for measuredValueType, measuredValues := range splittedMeasuredValues {
var queryFile string
switch measuredValueType {
case types.Humidity:
queryFile = postgres.InsertOrUpdateHumiditySQL
case types.Pressure:
queryFile = postgres.InsertOrUpdatePressureSQL
case types.Temperature:
queryFile = postgres.InsertOrUpdateTemperatureSQL
default:
return fmt.Errorf("Measured value type %v not supported", measuredValueType)
}
err := d.insertOrUpdateMeasuredValues(tx, queryFile, measuredValues...)
if err != nil {
return err
}
}
return tx.Commit()
}
func (d *Postgres) insertOrUpdateMeasuredValues(tx *sql.Tx, query string, measuredValues ...*types.MeasuredValue) error {
stmt, err := tx.Prepare(query)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, measuredValue := range measuredValues {
if measuredValue.CreationDate.Equal(time.Time{}) {
measuredValue.CreationDate = time.Now()
}
_, err := stmt.Exec(
&measuredValue.ID,
&measuredValue.Value,
&measuredValue.Date,
&measuredValue.SensorID,
&measuredValue.CreationDate,
&measuredValue.UpdateDate,
)
if err != nil {
return fmt.Errorf("Failed to execute statement: %v", err)
}
}
return nil
}
// AddSensors into the database
func (d *Postgres) AddSensors(ctx context.Context, sensors ...*types.Sensor) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
stmt, err := tx.Prepare(postgres.InsertSensorSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, sensor := range sensors {
if sensor.CreationDate.Equal(time.Time{}) {
sensor.CreationDate = time.Now()
}
_, err = stmt.Exec(
&sensor.ID,
&sensor.Name,
&sensor.Location,
&sensor.WireID,
&sensor.I2CBus,
&sensor.I2CAddress,
&sensor.GPIONumber,
&sensor.Model,
&sensor.Enabled,
&sensor.TickDuration,
&sensor.DeviceID,
&sensor.CreationDate,
&sensor.UpdateDate,
)
if err != nil {
return fmt.Errorf("Failed to execute statement: %v", err)
}
}
return tx.Commit()
}
// AddOrUpdateSensors into the database
func (d *Postgres) AddOrUpdateSensors(ctx context.Context, sensors ...*types.Sensor) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
err = d.insertOrUpdateSensors(tx, sensors...)
if err != nil {
return err
}
return tx.Commit()
}
func (d *Postgres) insertOrUpdateSensors(tx *sql.Tx, sensors ...*types.Sensor) error {
stmt, err := tx.Prepare(postgres.InsertOrUpdateSensorSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, sensor := range sensors {
if sensor.CreationDate.Equal(time.Time{}) {
sensor.CreationDate = time.Now()
}
_, err = stmt.Exec(
&sensor.ID,
&sensor.Name,
&sensor.Location,
&sensor.WireID,
&sensor.I2CBus,
&sensor.I2CAddress,
&sensor.GPIONumber,
&sensor.Model,
&sensor.Enabled,
&sensor.TickDuration,
&sensor.DeviceID,
&sensor.CreationDate,
&sensor.UpdateDate,
)
if err != nil {
return fmt.Errorf("Failed to execute statement: %v", err)
}
}
return nil
}
// Close closes the database and prevents new queries from starting. Close then
// waits for all queries that have started processing on the server to finish.
func (postgres *Postgres) Close() error {
return postgres.dbo.Close()
}
// Migrate creates all required tables if not exist
func (d *Postgres) Migrate(ctx context.Context) error {
sourceDriver, err := iofs.New(postgres.DDLAssets, postgres.DDLAssetPath)
if err != nil {
return err
}
m, err := migrate.NewWithSourceInstance("iofs", sourceDriver, d.databaseURL.String())
if err != nil {
return err
}
err = m.Up()
switch {
case errors.Is(err, migrate.ErrNoChange):
return nil
default:
return err
}
}
// GetDevice from database
func (d *Postgres) GetDeviceByID(ctx context.Context, id string) (*types.Device, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
devices, err := d.selectDevices(tx, postgres.SelectDeviceByIDSQL, id)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
if len(devices) == 0 {
return nil, nil
}
return devices[0], nil
}
// GetDevice from database
func (d *Postgres) GetDeviceByName(ctx context.Context, name string) (*types.Device, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
devices, err := d.selectDevices(tx, postgres.SelectDeviceByNameSQL, name)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
if len(devices) == 0 {
return nil, nil
}
return devices[0], nil
}
// GetDevices from the database
func (d *Postgres) GetDevices(ctx context.Context) ([]*types.Device, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
devices, err := d.selectDevices(tx, postgres.SelectDevicesSQL)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
}
return devices, nil
}
func (d *Postgres) selectDevices(tx *sql.Tx, query string, args ...interface{}) ([]*types.Device, error) {
stmt, err := tx.Prepare(query)
if err != nil {
return nil, fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
rows, err := stmt.Query(args...)
if err != nil {
return nil, fmt.Errorf("Failed to query statement: %v", err)
}
defer func() { _ = rows.Close() }()
devices := make([]*types.Device, 0)
for rows.Next() {
device := new(types.Device)
err = rows.Scan(
&device.ID,
&device.Name,
&device.Location,
&device.CreationDate,
&device.UpdateDate,
)
if err != nil {
return nil, fmt.Errorf("Failed to scan row: %v", err)
}
devices = append(devices, device)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("failed to scan rows: %v", err)
}
return devices, nil
}
// GetHumidity returns humidity from the database
func (d *Postgres) GetHumidityByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
measuredValues, err := d.selectMeasuredValue(tx, postgres.SelectHumidityByIDSQL, id)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
if measuredValues == nil {
return nil, nil
}
for _, measuredValue := range measuredValues {
measuredValue.ValueType = types.Humidity
}
return measuredValues[0], nil
}
// GetHumidities returns humidities from the database
func (d *Postgres) GetHumidities(ctx context.Context) ([]*types.MeasuredValue, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
measuredValues, err := d.selectMeasuredValue(tx, postgres.SelectHumiditiesSQL)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
for _, measuredValue := range measuredValues {
measuredValue.ValueType = types.Humidity
}
return measuredValues, nil
}
func (d *Postgres) selectMeasuredValue(tx *sql.Tx, query string, args ...interface{}) ([]*types.MeasuredValue, error) {
stmt, err := tx.Prepare(query)
if err != nil {
return nil, err
}
defer func() { _ = stmt.Close() }()
rows, err := stmt.Query(args...)
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
measuredValues := make([]*types.MeasuredValue, 0)
for rows.Next() {
measuredValue := new(types.MeasuredValue)
err := rows.Scan(
&measuredValue.ID,
&measuredValue.Value,
&measuredValue.Date,
&measuredValue.SensorID,
&measuredValue.CreationDate,
&measuredValue.UpdateDate,
)
if err != nil {
return nil, fmt.Errorf("failed to scan row: %v", err)
}
measuredValues = append(measuredValues, measuredValue)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("failed to scan rows: %v", err)
}
return measuredValues, nil
}
// GetPressure returns pressure from the database
func (d *Postgres) GetPressureByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
measuredValues, err := d.selectMeasuredValue(tx, postgres.SelectPressureByIDSQL, id)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
if measuredValues == nil {
return nil, nil
}
for _, measuredValue := range measuredValues {
measuredValue.ValueType = types.Pressure
}
return measuredValues[0], nil
}
// GetPressures returns pressure from the database
func (d *Postgres) GetPressures(ctx context.Context) ([]*types.MeasuredValue, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
measuredValues, err := d.selectMeasuredValue(tx, postgres.SelectPressuresSQL)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
for _, measuredValue := range measuredValues {
measuredValue.ValueType = types.Pressure
}
return measuredValues, nil
}
// GetSensor from database
func (d *Postgres) GetSensorByID(ctx context.Context, id string) (*types.Sensor, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
sensors, err := d.selectSensors(tx, postgres.SelectSensorByIDSQL, id)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
}
if len(sensors) == 0 {
return nil, nil
}
return sensors[0], nil
}
// GetSensors from the database
func (d *Postgres) GetSensors(ctx context.Context) ([]*types.Sensor, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
sensors, err := d.selectSensors(tx, postgres.SelectSensorsSQL)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
}
return sensors, nil
}
// GetSensorsByModels from the database
func (d *Postgres) GetSensorsByDeviceIDs(ctx context.Context, deviceIDs ...string) ([]*types.Sensor, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
cachedSensors := make([]*types.Sensor, 0)
for i := range deviceIDs {
sensors, err := d.selectSensors(tx, postgres.SelectSensorsByDeviceIDSQL, deviceIDs[i])
if err != nil {
return nil, err
}
cachedSensors = append(cachedSensors, sensors...)
}
err = tx.Commit()
if err != nil {
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
}
return cachedSensors, nil
}
// GetSensorsByModel from the database
func (d *Postgres) GetSensorsByModels(ctx context.Context, sensorModels ...string) ([]*types.Sensor, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
cachedSensors := make([]*types.Sensor, 0)
for i := range sensorModels {
sensors, err := d.selectSensors(tx, postgres.SelectSensorsByModelSQL, sensorModels[i])
if err != nil {
return nil, err
}
cachedSensors = append(cachedSensors, sensors...)
}
err = tx.Commit()
if err != nil {
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
}
return cachedSensors, nil
}
// GetSensorsByModel from the database
func (d *Postgres) GetSensorsByNames(ctx context.Context, sensorNames ...string) ([]*types.Sensor, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
cachedSensors := make([]*types.Sensor, 0)
for i := range sensorNames {
sensors, err := d.selectSensors(tx, postgres.SelectSensorsByNameSQL, sensorNames[i])
if err != nil {
return nil, err
}
cachedSensors = append(cachedSensors, sensors...)
}
err = tx.Commit()
if err != nil {
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
}
return cachedSensors, nil
}
func (d *Postgres) selectSensors(tx *sql.Tx, query string, args ...interface{}) ([]*types.Sensor, error) {
stmt, err := tx.Prepare(query)
if err != nil {
return nil, fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
rows, err := stmt.Query(args...)
if err != nil {
return nil, fmt.Errorf("Failed to query statement: %v", err)
}
defer func() { _ = rows.Close() }()
sensors := make([]*types.Sensor, 0)
for rows.Next() {
sensor := new(types.Sensor)
err = rows.Scan(
&sensor.ID,
&sensor.Name,
&sensor.Location,
&sensor.WireID,
&sensor.I2CBus,
&sensor.I2CAddress,
&sensor.GPIONumber,
&sensor.Model,
&sensor.Enabled,
&sensor.TickDuration,
&sensor.DeviceID,
&sensor.CreationDate,
&sensor.UpdateDate,
)
if err != nil {
return nil, fmt.Errorf("Failed to scan row: %v", err)
}
sensors = append(sensors, sensor)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("failed to scan rows: %v", err)
}
return sensors, nil
}
// GetTemperature returns temperatures from the database
func (d *Postgres) GetTemperatureByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
measuredValues, err := d.selectMeasuredValue(tx, postgres.SelectTemperatureByIDSQL, id)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
if measuredValues == nil {
return nil, nil
}
for _, measuredValue := range measuredValues {
measuredValue.ValueType = types.Temperature
}
return measuredValues[0], nil
}
// GetTemperatures returns temperatures from the database
func (d *Postgres) GetTemperatures(ctx context.Context) ([]*types.MeasuredValue, error) {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
measuredValues, err := d.selectMeasuredValue(tx, postgres.SelectTemperaturesSQL)
if err != nil {
return nil, err
}
err = tx.Commit()
if err != nil {
return nil, err
}
for _, measuredValue := range measuredValues {
measuredValue.ValueType = types.Temperature
}
return measuredValues, nil
}
// Import imports devices, sensors and all measured values from a source
// repository. Existing entries will be updated.
func (d *Postgres) Import(ctx context.Context, src Repository) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
devices, err := src.GetDevices(ctx)
if err != nil {
return err
}
err = d.insertOrUpdateDevices(tx, devices...)
if err != nil {
return err
}
sensors, err := src.GetSensors(ctx)
if err != nil {
return err
}
err = d.insertOrUpdateSensors(tx, sensors...)
if err != nil {
return err
}
for _, value := range []struct {
f func(ctx context.Context) ([]*types.MeasuredValue, error)
query string
}{
{f: src.GetHumidities, query: postgres.InsertOrUpdateHumiditySQL},
{f: src.GetPressures, query: postgres.InsertOrUpdatePressureSQL},
{f: src.GetTemperatures, query: postgres.InsertOrUpdateTemperatureSQL},
} {
measuredValues, err := value.f(ctx)
if err != nil {
return err
}
err = d.insertOrUpdateMeasuredValues(tx, value.query, measuredValues...)
if err != nil {
return err
}
}
return tx.Commit()
}
// RemoveDevices from the database
func (d *Postgres) RemoveDevicesByIDs(ctx context.Context, deviceIDs ...string) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
stmt, err := tx.Prepare(postgres.DeleteDeviceByIDSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, deviceID := range deviceIDs {
_, err = stmt.Exec(deviceID)
if err != nil {
return err
}
}
return tx.Commit()
}
// RemoveDevices from the database
func (d *Postgres) RemoveDevicesByNames(ctx context.Context, names ...string) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
stmt, err := tx.Prepare(postgres.DeleteDeviceByNameSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, deviceID := range names {
_, err = stmt.Exec(deviceID)
if err != nil {
return err
}
}
return tx.Commit()
}
// RemoveSensors from the database
func (d *Postgres) RemoveSensorsByIDs(ctx context.Context, sensorIDs ...string) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
stmt, err := tx.Prepare(postgres.DeleteSensorByIDSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, sensorID := range sensorIDs {
_, err = stmt.Exec(sensorID)
if err != nil {
return err
}
}
return tx.Commit()
}
// RemoveSensors from the database
func (d *Postgres) RemoveSensorsByNames(ctx context.Context, sensorIDs ...string) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return fmt.Errorf("Failed to begin new transaction: %v", err)
}
defer func() { _ = tx.Rollback() }()
stmt, err := tx.Prepare(postgres.DeleteSensorByNameSQL)
if err != nil {
return fmt.Errorf("Failed to prepare statement: %v", err)
}
defer func() { _ = stmt.Close() }()
for _, sensorID := range sensorIDs {
_, err = stmt.Exec(sensorID)
if err != nil {
return err
}
}
return tx.Commit()
}
// UpdateDevices updates a device in the database
func (d *Postgres) UpdateDevices(ctx context.Context, devices ...*types.Device) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
stmt, err := tx.Prepare(postgres.UpdateDeviceSQL)
if err != nil {
return err
}
defer func() { _ = stmt.Close() }()
for _, device := range devices {
now := time.Now()
device.UpdateDate = &now
_, err := stmt.Exec(
&device.Name,
&device.Location,
&device.CreationDate,
&device.UpdateDate,
&device.ID,
)
if err != nil {
return err
}
}
return tx.Commit()
}
// UpdateSensors updates a sensor in the database
func (d *Postgres) UpdateSensors(ctx context.Context, sensors ...*types.Sensor) error {
tx, err := d.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
stmt, err := tx.Prepare(postgres.UpdateSensorSQL)
if err != nil {
return err
}
defer func() { _ = stmt.Close() }()
for _, sensor := range sensors {
now := time.Now()
sensor.UpdateDate = &now
_, err := stmt.Exec(
&sensor.Name,
&sensor.Location,
&sensor.WireID,
&sensor.I2CBus,
&sensor.I2CAddress,
&sensor.GPIONumber,
&sensor.Model,
&sensor.Enabled,
&sensor.TickDuration,
&sensor.DeviceID,
&sensor.CreationDate,
&sensor.UpdateDate,
&sensor.ID,
)
if err != nil {
return err
}
}
return tx.Commit()
}