Add a data connector for Attio CRM

Author: exAspArk

.env.sample

@@ -10,5 +10,7 @@ SOURCE_POSTGRES_DATABASE_URL=postgres://postgres:postgres@host.docker.internal:5
 SOURCE_AMPLITUDE_API_KEY=[REPLACE_ME]
 SOURCE_AMPLITUDE_SECRET_KEY=[REPLACE_ME]
 
+SOURCE_ATTIO_API_ACCESS_TOKEN=[REPLACE_ME]
+
 BEMIDB_LOG_LEVEL=INFO
 BEMIDB_DISABLE_ANONYMOUS_ANALYTICS=true

Dockerfile

@@ -43,24 +43,34 @@ RUN cd /app/src/syncer-postgres && go mod download
 COPY --chown=app:app src/syncer-amplitude/go.mod src/syncer-amplitude/go.sum /app/src/syncer-amplitude/
 RUN cd /app/src/syncer-amplitude && go mod download
 
+COPY --chown=app:app src/syncer-attio/go.mod src/syncer-attio/go.sum /app/src/syncer-attio/
+RUN cd /app/src/syncer-attio && go mod download
+
 COPY --chown=app:app src/server/go.mod src/server/go.sum /app/src/server/
 RUN cd /app/src/server && go mod download
 
 COPY --chown=app:app src/common /app/src/common
 COPY --chown=app:app src/syncer-postgres /app/src/syncer-postgres
 COPY --chown=app:app src/syncer-amplitude /app/src/syncer-amplitude
+COPY --chown=app:app src/syncer-attio /app/src/syncer-attio
 COPY --chown=app:app src/server /app/src/server
 
 RUN ARCH=$(dpkg --print-architecture) \
   && cd /app/src/syncer-postgres && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/syncer-postgres \
     && cd /app/src/syncer-amplitude && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/syncer-amplitude \
+    && cd /app/src/syncer-attio && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/syncer-attio \
     && cd /app/src/server && CGO_ENABLED=1 GOOS=linux GOARCH=$ARCH go build -o /app/bin/server
 
 # Prepare final image ##############################################################################
 
 FROM base AS final
 
-COPY --chown=app:app --from=compile /app/bin/syncer-postgres /app/bin/syncer-amplitude /app/bin/server /app/bin/
+COPY --chown=app:app --from=compile \
+  /app/bin/syncer-postgres \
+  /app/bin/syncer-amplitude \
+  /app/bin/syncer-attio \
+  /app/bin/server \
+  /app/bin/
 COPY --chown=app:app docker/bin /app/bin/
 
 USER app

Makefile

@@ -5,12 +5,14 @@ install:
 	devbox run "cd src/common && go mod tidy && \
 		cd ../syncer-postgres && go mod tidy && \
 		cd ../syncer-amplitude && go mod tidy && \
+		cd ../syncer-attio && go mod tidy && \
 		cd ../server && go mod tidy"
 
 lint:
 	devbox run "cd src/common && go fmt && staticcheck . && \
 		cd ../syncer-postgres && go fmt && deadcode . && staticcheck . && \
 		cd ../syncer-amplitude && go fmt && deadcode . && staticcheck . && \
+		cd ../syncer-attio && go fmt && deadcode . && staticcheck . && \
 		cd ../server && go fmt && deadcode . && staticcheck ."
 
 build:
@@ -37,6 +39,9 @@ local-syncer-postgres: local-build
 local-syncer-amplitude: local-build
 	docker run -it --rm --env-file .env -e DESTINATION_SCHEMA_NAME=amplitude bemidb:local syncer-amplitude
 
+local-syncer-attio: local-build
+	docker run -it --rm --env-file .env -e DESTINATION_SCHEMA_NAME=attio bemidb:local syncer-attio
+
 local-sh:
 	docker run -it --rm --env-file .env bemidb:local bash
 

README.md

@@ -165,6 +165,16 @@ docker run \
   ghcr.io/bemihq/bemidb:latest syncer-amplitude
 ```
 
+#### Syncing from Attio
+
+```sh
+docker run \
+  -e SOURCE_ATTIO_API_ACCESS_TOKEN=[...] \
+  -e DESTINATION_SCHEMA_NAME=attio \
+  -e AWS_REGION -e AWS_S3_BUCKET -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY -e CATALOG_DATABASE_URL \
+  ghcr.io/bemihq/bemidb:latest syncer-attio
+```
+
 #### Customizing S3 endpoint
 
 BemiDB can work with various S3-compatible object storage solutions, such as MinIO.
@@ -215,6 +225,13 @@ export AWS_S3_ENDPOINT=http://localhost:9000
 | `SOURCE_AMPLITUDE_SECRET_KEY` | Required      | Amplitude secret key for authentication.                           |
 | `SOURCE_AMPLITUDE_START_DATE` | `2025-01-01`  | Start date for syncing data from Amplitude in `YYYY-MM-DD` format. |
 
+#### `syncer-attio` command options
+
+| Environment variable            | Default value | Description                                |
+|---------------------------------|---------------|--------------------------------------------|
+| `DESTINATION_SCHEMA_NAME`       | Required      | Schema name in BemiDB to sync data to.     |
+| `SOURCE_ATTIO_API_ACCESS_TOKEN` | Required      | Attio API access token for authentication. |
+
 #### `server` command options
 
 | Environment variable | Default value | Description                            |
@@ -314,6 +331,7 @@ SELECT * FROM [TABLE] WHERE [JSON_COLUMN]->>'[JSON_KEY]' = '[JSON_VALUE]';
   - [ ] Jupyter notebooks ([#27](https://github.com/BemiHQ/BemiDB/issues/27))
 - [x] Data syncing from other sources
   - [x] Amplitude (incremental)
+  - [x] Attio CRM (full-refresh)
   - [x] Postgres (full-refresh)
   - [ ] HubSpot
   - [ ] Stripe

docker/bin/run.sh

@@ -27,6 +27,16 @@ case "${1:-}" in
     ./bin/syncer-amplitude 2>&1 | sed 's/^/[Syncer] /'
     echo "Syncer for Amplitude finished."
     ;;
+  syncer-attio)
+    : "${SOURCE_ATTIO_API_ACCESS_TOKEN:?Environment variable SOURCE_ATTIO_API_ACCESS_TOKEN must be set}"
+    : "${DESTINATION_SCHEMA_NAME:?Environment variable DESTINATION_SCHEMA_NAME must be set}"
+
+    psql $CATALOG_DATABASE_URL -f /app/scripts/catalog.sql
+
+    echo "Starting Syncer for Attio..."
+    ./bin/syncer-attio 2>&1 | sed 's/^/[Syncer] /'
+    echo "Syncer for Attio finished."
+    ;;
   server)
     : "${AWS_REGION:?Environment variable AWS_REGION must be set}"
     : "${AWS_S3_BUCKET:?Environment variable AWS_S3_BUCKET must be set}"

src/common/capped_buffer.go

@@ -6,6 +6,8 @@ import (
 	"sync"
 )
 
+const DEFAULT_CAPPED_BUFFER_SIZE = 32 * 1024 * 1024 // 32 MB in memory
+
 type CappedBuffer struct {
 	Config       *CommonConfig
 	MaxSizeBytes int

src/common/common_config.go

@@ -1,7 +1,7 @@
 package common
 
 const (
-	VERSION = "1.2.0"
+	VERSION = "1.3.0"
 
 	ENV_LOG_LEVEL                   = "BEMIDB_LOG_LEVEL"
 	ENV_DISABLE_ANONYMOUS_ANALYTICS = "BEMIDB_DISABLE_ANONYMOUS_ANALYTICS"

src/common/iceberg_schema_column.go

@@ -368,8 +368,17 @@ func (col *IcebergSchemaColumn) duckdbPrimitiveValueFromJson(value any) interfac
 			return value
 		}
 	case IcebergColumnTypeDate:
-		days := value.(float64)
-		return time.Unix(0, 0).UTC().AddDate(0, 0, int(days))
+		switch kind {
+		case reflect.String:
+			valueString := value.(string)
+			if valueString == "" {
+				return nil
+			}
+			return StringDateToTime(valueString)
+		case reflect.Float64:
+			days := value.(float64)
+			return time.Unix(0, 0).UTC().AddDate(0, 0, int(days))
+		}
 	case IcebergColumnTypeTime:
 		var nanoseconds int64
 		if col.DatetimePrecision == 6 {

src/common/iceberg_table.go

@@ -43,6 +43,32 @@ func (table *IcebergTable) Create(tableS3Path string) {
 	table.IcebergCatalog.CreateTable(table.IcebergSchemaTable, tableS3Path+"/metadata/"+ICEBERG_METADATA_INITIAL_FILE_NAME)
 }
 
+func (table *IcebergTable) ReplaceWith(callbackFunc func(syncingIcebergTable *IcebergTable)) {
+	originalTableName := table.IcebergSchemaTable.Table
+
+	// Delete -syncing table
+	syncingIcebergSchemaTable := IcebergSchemaTable{Schema: table.IcebergSchemaTable.Schema, Table: originalTableName + TEMP_TABLE_SUFFIX_SYNCING}
+	syncingIcebergTable := NewIcebergTable(table.Config, table.StorageS3, table.DuckdbClient, syncingIcebergSchemaTable)
+	syncingIcebergTable.DropIfExists()
+
+	// Insert into -syncing table
+	callbackFunc(syncingIcebergTable)
+
+	// Delete -deleting table
+	deletingIcebergSchemaTable := IcebergSchemaTable{Schema: table.IcebergSchemaTable.Schema, Table: originalTableName + TEMP_TABLE_SUFFIX_DELETING}
+	deletingIcebergTable := NewIcebergTable(table.Config, table.StorageS3, table.DuckdbClient, deletingIcebergSchemaTable)
+	deletingIcebergTable.DropIfExists()
+
+	// Rename table to -deleting
+	table.Rename(deletingIcebergSchemaTable.Table)
+
+	// Rename -syncing to table
+	syncingIcebergTable.Rename(originalTableName)
+
+	// Delete -deleting table
+	deletingIcebergTable.DropIfExists()
+}
+
 func (table *IcebergTable) DropIfExists() {
 	tableS3Path := table.IcebergCatalog.TableS3Path(table.IcebergSchemaTable)
 	if tableS3Path == "" {

src/syncer-amplitude/event.go

@@ -128,7 +128,7 @@ func (event *Event) ToMap() map[string]interface{} {
 	return result
 }
 
-func EventIcebergSchemaColumns(config *common.CommonConfig) []*common.IcebergSchemaColumn {
+func EventsIcebergSchemaColumns(config *common.CommonConfig) []*common.IcebergSchemaColumn {
 	return []*common.IcebergSchemaColumn{
 		{Config: config, ColumnName: "adid", ColumnType: common.IcebergColumnTypeString, Position: 1},
 		{Config: config, ColumnName: "amplitude_attribution_ids", ColumnType: common.IcebergColumnTypeString, Position: 2},