411 lines
13 KiB
Go
411 lines
13 KiB
Go
|
package db
|
||
|
|
||
|
import (
|
||
|
"context"
|
||
|
"database/sql"
|
||
|
"fmt"
|
||
|
|
||
|
"github.com/go-flucky/flucky/pkg/types"
|
||
|
_ "github.com/lib/pq"
|
||
|
)
|
||
|
|
||
|
type Postgres struct {
|
||
|
dbo *sql.DB
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) Close() error {
|
||
|
return p.Close()
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) DeleteDevices(ctx context.Context, devices []*types.Device) error {
|
||
|
query := string(MustAsset("pkg/db/sql/psql/deleteDevice.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, device := range devices {
|
||
|
_, err := stmt.ExecContext(ctx, &device.DeviceID)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) DeleteSensors(ctx context.Context, sensors []*types.Sensor) error {
|
||
|
query := string(MustAsset("pkg/db/sql/psql/deleteSensor.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, sensor := range sensors {
|
||
|
_, err := stmt.ExecContext(ctx, &sensor.SensorID)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) DeleteMeasuredValues(ctx context.Context, measuredValues []*types.MeasuredValue) error {
|
||
|
|
||
|
deleteMeasuredValue := func(ctx context.Context, query string, measuredValues []*types.MeasuredValue) error {
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, measuredValue := range measuredValues {
|
||
|
_, err := stmt.ExecContext(ctx, &measuredValue.ID)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
sortedMeasuredValueTypes := make(map[types.MeasuredValueType][]*types.MeasuredValue)
|
||
|
for _, measuredValue := range measuredValues {
|
||
|
if _, ok := sortedMeasuredValueTypes[measuredValue.ValueType]; !ok {
|
||
|
sortedMeasuredValueTypes[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
|
||
|
}
|
||
|
sortedMeasuredValueTypes[measuredValue.ValueType] = append(sortedMeasuredValueTypes[measuredValue.ValueType], measuredValue)
|
||
|
}
|
||
|
|
||
|
for measuredValueType, sortedMeasuredValues := range sortedMeasuredValueTypes {
|
||
|
switch measuredValueType {
|
||
|
case types.MeasuredValueTypeHumidity:
|
||
|
if err := deleteMeasuredValue(ctx, string(MustAsset("pkg/db/sql/psql/deleteHumidity.sql")), sortedMeasuredValues); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
case types.MeasuredValueTypePressure:
|
||
|
if err := deleteMeasuredValue(ctx, string(MustAsset("pkg/db/sql/psql/deletePressure.sql")), sortedMeasuredValues); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
case types.MeasuredValueTypeTemperature:
|
||
|
if err := deleteMeasuredValue(ctx, string(MustAsset("pkg/db/sql/psql/deleteTemperature.sql")), sortedMeasuredValues); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) InsertDevices(ctx context.Context, devices []*types.Device) error {
|
||
|
|
||
|
query := string(MustAsset("pkg/db/sql/psql/insertDevice.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, device := range devices {
|
||
|
_, err := stmt.ExecContext(ctx, &device.DeviceID, &device.DeviceName, &device.DeviceLocation, &device.CreationDate)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) InsertMeasuredValues(ctx context.Context, measuredValues []*types.MeasuredValue) error {
|
||
|
|
||
|
sortedMeasuredValueTypes := make(map[types.MeasuredValueType][]*types.MeasuredValue)
|
||
|
|
||
|
for _, measuredValue := range measuredValues {
|
||
|
if _, ok := sortedMeasuredValueTypes[measuredValue.ValueType]; !ok {
|
||
|
sortedMeasuredValueTypes[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
|
||
|
}
|
||
|
sortedMeasuredValueTypes[measuredValue.ValueType] = append(sortedMeasuredValueTypes[measuredValue.ValueType], measuredValue)
|
||
|
}
|
||
|
|
||
|
for measuredValueType, sortedMeasuredValues := range sortedMeasuredValueTypes {
|
||
|
switch measuredValueType {
|
||
|
case types.MeasuredValueTypeHumidity:
|
||
|
if err := p.insertHumidity(ctx, sortedMeasuredValues); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
case types.MeasuredValueTypePressure:
|
||
|
if err := p.insertPressure(ctx, sortedMeasuredValues); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
case types.MeasuredValueTypeTemperature:
|
||
|
if err := p.insertTemperature(ctx, sortedMeasuredValues); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) insertHumidity(ctx context.Context, measuredValues []*types.MeasuredValue) error {
|
||
|
|
||
|
query := string(MustAsset("pkg/db/sql/psql/insertHumidity.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, measuredValue := range measuredValues {
|
||
|
|
||
|
if measuredValue.ValueType != types.MeasuredValueTypeHumidity {
|
||
|
continue
|
||
|
}
|
||
|
|
||
|
_, err := stmt.ExecContext(ctx, &measuredValue.ID, &measuredValue.Value, &measuredValue.FromDate, &measuredValue.TillDate, &measuredValue.SensorID, &measuredValue.CreationDate, &measuredValue.UpdateDate)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) insertPressure(ctx context.Context, measuredValues []*types.MeasuredValue) error {
|
||
|
|
||
|
query := string(MustAsset("pkg/db/sql/psql/insertPressure.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, measuredValue := range measuredValues {
|
||
|
|
||
|
if measuredValue.ValueType != types.MeasuredValueTypePressure {
|
||
|
continue
|
||
|
}
|
||
|
|
||
|
_, err := stmt.ExecContext(ctx, &measuredValue.ID, &measuredValue.Value, &measuredValue.FromDate, &measuredValue.TillDate, &measuredValue.SensorID, &measuredValue.CreationDate, &measuredValue.UpdateDate)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) insertTemperature(ctx context.Context, measuredValues []*types.MeasuredValue) error {
|
||
|
|
||
|
query := string(MustAsset("pkg/db/sql/psql/insertTemperature.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, measuredValue := range measuredValues {
|
||
|
|
||
|
if measuredValue.ValueType != types.MeasuredValueTypeTemperature {
|
||
|
continue
|
||
|
}
|
||
|
|
||
|
_, err := stmt.ExecContext(ctx, &measuredValue.ID, &measuredValue.Value, &measuredValue.FromDate, &measuredValue.TillDate, &measuredValue.SensorID, &measuredValue.CreationDate, &measuredValue.UpdateDate)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) InsertSensors(ctx context.Context, sensors []*types.Sensor) error {
|
||
|
|
||
|
query := string(MustAsset("pkg/db/sql/psql/insertSensor.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
defer stmt.Close()
|
||
|
|
||
|
for _, sensor := range sensors {
|
||
|
|
||
|
_, err := stmt.ExecContext(ctx, &sensor.SensorID, &sensor.SensorName, &sensor.SensorLocation, &sensor.WireID, &sensor.I2CBus, &sensor.I2CAddress, &sensor.GPIONumber, &sensor.SensorModel, &sensor.SensorEnabled, &sensor.DeviceID, &sensor.CreationDate)
|
||
|
if err != nil {
|
||
|
return fmt.Errorf("%v: %v", errorStatementExecute, err)
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectDeviceByID(ctx context.Context, id string) (*types.Device, error) {
|
||
|
query := string(MustAsset("pkg/db/sql/psql/selectDeviceByID.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return nil, fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
|
||
|
row := stmt.QueryRowContext(ctx, id)
|
||
|
if row == nil {
|
||
|
return nil, errorRowNotFound
|
||
|
}
|
||
|
|
||
|
device := new(types.Device)
|
||
|
err = row.Scan(&device.DeviceID, &device.DeviceName, &device.DeviceLocation, &device.CreationDate)
|
||
|
if err != nil {
|
||
|
return nil, fmt.Errorf("%v: %v", errorScanRow, err)
|
||
|
}
|
||
|
|
||
|
return device, nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectMeasuredValuesByIDAndType(ctx context.Context, id string, valueType types.MeasuredValueType) (*types.MeasuredValue, error) {
|
||
|
switch valueType {
|
||
|
case types.MeasuredValueTypeHumidity:
|
||
|
return p.SelectHumidityByID(ctx, id)
|
||
|
case types.MeasuredValueTypePressure:
|
||
|
return p.SelectPressureByID(ctx, id)
|
||
|
case types.MeasuredValueTypeTemperature:
|
||
|
return p.SelectTemperatureByID(ctx, id)
|
||
|
default:
|
||
|
return nil, fmt.Errorf("%v: %v", errorUnknownMeasuredValueType, valueType)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectHumidities(ctx context.Context) ([]*types.MeasuredValue, error) {
|
||
|
queryFile := "pkg/db/sql/psql/selectHumidities.sql"
|
||
|
measuredValues, err := p.selectMeasuredValues(ctx, types.MeasuredValueTypeHumidity, queryFile, nil)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
return measuredValues, nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectHumidityByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
||
|
queryFile := "pkg/db/sql/psql/selectHumidityByID.sql"
|
||
|
args := []interface{}{id}
|
||
|
measuredValues, err := p.selectMeasuredValues(ctx, types.MeasuredValueTypeHumidity, queryFile, args)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
if len(measuredValues) == 0 {
|
||
|
return nil, fmt.Errorf("%v: %v", errorRowNotFound, id)
|
||
|
}
|
||
|
|
||
|
return measuredValues[0], nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectPressures(ctx context.Context) ([]*types.MeasuredValue, error) {
|
||
|
queryFile := "pkg/db/sql/psql/selectPressures.sql"
|
||
|
measuredValues, err := p.selectMeasuredValues(ctx, types.MeasuredValueTypePressure, queryFile, nil)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
return measuredValues, nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectPressureByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
||
|
queryFile := "pkg/db/sql/psql/selectPressureByID.sql"
|
||
|
args := []interface{}{id}
|
||
|
measuredValues, err := p.selectMeasuredValues(ctx, types.MeasuredValueTypePressure, queryFile, args)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
if len(measuredValues) == 0 {
|
||
|
return nil, fmt.Errorf("%v: %v", errorRowNotFound, id)
|
||
|
}
|
||
|
|
||
|
return measuredValues[0], nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectSensorByID(ctx context.Context, id string) (*types.Sensor, error) {
|
||
|
query := string(MustAsset("pkg/db/sql/psql/selectSensorByID.sql"))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return nil, fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
|
||
|
row := stmt.QueryRowContext(ctx, id)
|
||
|
if row == nil {
|
||
|
return nil, errorRowNotFound
|
||
|
}
|
||
|
|
||
|
sensor := new(types.Sensor)
|
||
|
err = row.Scan(&sensor.SensorID, &sensor.SensorName, &sensor.SensorLocation, &sensor.WireID, &sensor.I2CBus, &sensor.I2CAddress, &sensor.GPIONumber, &sensor.SensorModel, &sensor.SensorEnabled, &sensor.DeviceID, &sensor.CreationDate)
|
||
|
if err != nil {
|
||
|
return nil, fmt.Errorf("%v: %v", errorScanRow, err)
|
||
|
}
|
||
|
|
||
|
return sensor, nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectTemperatures(ctx context.Context) ([]*types.MeasuredValue, error) {
|
||
|
queryFile := "pkg/db/sql/psql/selectTemperatures.sql"
|
||
|
measuredValues, err := p.selectMeasuredValues(ctx, types.MeasuredValueTypeTemperature, queryFile, nil)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
return measuredValues, nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) SelectTemperatureByID(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
||
|
queryFile := "pkg/db/sql/psql/selectTemperatureByID.sql"
|
||
|
args := []interface{}{id}
|
||
|
measuredValues, err := p.selectMeasuredValues(ctx, types.MeasuredValueTypeTemperature, queryFile, args)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
if len(measuredValues) == 0 {
|
||
|
return nil, fmt.Errorf("%v: %v", errorRowNotFound, id)
|
||
|
}
|
||
|
|
||
|
return measuredValues[0], nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) selectMeasuredValues(ctx context.Context, measuredValueType types.MeasuredValueType, queryFile string, queryArgs []interface{}) ([]*types.MeasuredValue, error) {
|
||
|
query := string(MustAsset(queryFile))
|
||
|
|
||
|
stmt, err := p.dbo.PrepareContext(ctx, query)
|
||
|
if err != nil {
|
||
|
return nil, fmt.Errorf("%v: %v", errorPrepareStatement, err)
|
||
|
}
|
||
|
|
||
|
rows, err := stmt.QueryContext(ctx, queryArgs...)
|
||
|
if err != nil {
|
||
|
return nil, fmt.Errorf("%v: %v", errorStatementQuery, err)
|
||
|
}
|
||
|
|
||
|
measuredValues := make([]*types.MeasuredValue, 0)
|
||
|
for rows.Next() {
|
||
|
measuredValue := new(types.MeasuredValue)
|
||
|
measuredValue.ValueType = measuredValueType
|
||
|
rows.Scan(&measuredValue.ID, &measuredValue.Value, &measuredValue.FromDate, &measuredValue.TillDate, &measuredValue.SensorID, &measuredValue.CreationDate, &measuredValue.UpdateDate)
|
||
|
measuredValues = append(measuredValues, measuredValue)
|
||
|
}
|
||
|
|
||
|
return measuredValues, nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) UpdateDevices(ctx context.Context, devices []*types.Device) error {
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) UpdateMeasuredValues(ctx context.Context, measuredValues []*types.MeasuredValue) error {
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (p *Postgres) UpdateSensors(ctx context.Context, sensots []*types.Sensor) error {
|
||
|
return nil
|
||
|
}
|