frx/kurious
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

implement webserver

Browse Source
This commit is contained in:
Aleksandr Trushkin
2023-12-17 21:21:13 +03:00
parent f60ebcfb36
commit 1d4e8e10fb
12 changed files with 499 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
internal/kurious/adapters/ydb_course_repository.go
View File

@ -24,6 +24,29 @@ import (
yc "github.com/ydb-platform/ydb-go-yc"
)
var coursesFields = []string{
"id",
"external_id",
"source_type",
"source_name",
"course_thematic",
"learning_type",
"organization_id",
"origin_link",
"image_link",
"name",
"description",
"full_price",
"discount",
"duration",
"starts_at",
"created_at",
"updated_at",
"deleted_at",
}
var coursesFieldsStr = strings.Join(coursesFields, ",")
const (
defaultShutdownTimeout = time.Second * 10
)
@ -87,59 +110,31 @@ func (r *ydbCourseRepository) List(
) (result domain.ListCoursesResult, err error) {
const limit = 1000
const queryName = "list"
// const query = `
// DECLARE $limit AS Int32;
// DECLARE $id AS Text;
// SELECT
// id,
// external_id,
// source_type,
// source_name,
// course_thematic,
// learning_type,
// organization_id,
// origin_link,
// image_link,
// name,
// description,
// full_price,
// discount,
// duration,
// starts_at,
// created_at,
// updated_at,
// deleted_at
// FROM
// courses
// WHERE
// id > $id
// ORDER BY id
// LIMIT $limit;`
//
const fields = `id, external_id, source_type, source_name, course_thematic, learning_type, organization_id, origin_link, image_link, name, description, full_price, discount, duration, starts_at, created_at, updated_at, deleted_at`
if params.Limit == 0 {
params.Limit = limit
}
qtParams := queryTemplateParams{
Fields: fields,
Fields: coursesFieldsStr,
Table: "courses",
Suffix: "ORDER BY id\nLIMIT $limit",
Declares: []queryTemplateDeclaration{{
Name: "limit",
Type: "Int32",
}, {
Name: "id",
Type: "Text",
}},
Declares: []queryTemplateDeclaration{
{
Name: "limit",
Type: "Int32",
},
{
Name: "id",
Type: "Text",
},
},
Conditions: []string{
"id > $id",
},
}
options := make([]table.ParameterOption, 0, 4)
appendParams := func(name string, value string) {
opts := make([]table.ParameterOption, 0, 4)
appendTextParam := func(name string, value string) {
if value == "" {
return
}
@ -151,18 +146,17 @@ func (r *ydbCourseRepository) List(
}
qtParams.Declares = append(qtParams.Declares, d)
qtParams.Conditions = append(qtParams.Conditions, d.Name+"="+d.Arg())
options = append(options, table.ValueParam(d.Arg(), ydbvalue))
opts = append(opts, table.ValueParam(d.Arg(), ydbvalue))
}
appendParams("course_thematic", params.CourseThematic)
appendParams("learning_type", params.LearningType)
appendTextParam("course_thematic", params.CourseThematic)
appendTextParam("learning_type", params.LearningType)
var sb strings.Builder
err = template.Must(template.New("").Parse(queryTemplateSelect)).Execute(&sb, qtParams)
query, err := qtParams.render()
if err != nil {
return result, fmt.Errorf("executing template: %w", err)
return result, fmt.Errorf("rendering: %w", err)
}
query := sb.String()
xcontext.LogInfo(ctx, r.log, "planning to run query", slog.String("query", query), slog.Any("opts", opts))
courses := make([]domain.Course, 0, 1_000)
readTx := table.TxControl(
@ -171,9 +165,12 @@ func (r *ydbCourseRepository) List(
),
table.CommitTx(),
)
xcontext.LogInfo(ctx, r.log, "executing do")
err = r.db.Table().Do(
ctx,
func(ctx context.Context, s table.Session) error {
xcontext.LogInfo(ctx, r.log, "inside do")
start := time.Now()
defer func() {
since := time.Since(start)
@ -185,47 +182,51 @@ func (r *ydbCourseRepository) List(
)
}()
var lastKnownID string
for {
queryParams := table.NewQueryParameters(
table.ValueParam("$limit", types.Int32Value(limit)),
table.ValueParam("$id", types.TextValue(lastKnownID)),
)
_, res, err := s.Execute(
ctx, readTx, query, queryParams,
options.WithCollectStatsModeBasic(),
)
if err != nil {
return fmt.Errorf("executing: %w", err)
}
queryParams := table.NewQueryParameters(opts...)
if !res.NextResultSet(ctx) || !res.HasNextRow() {
break
}
xcontext.LogDebug(ctx, r.log, "executing")
for res.NextRow() {
var cdb courseDB
err = res.ScanNamed(cdb.getNamedValues()...)
if err != nil {
return fmt.Errorf("scanning row: %w", err)
}
courses = append(courses, mapCourseDB(cdb))
}
if err = res.Err(); err != nil {
return err
}
lastKnownID = courses[len(courses)-1].ID
_, res, err := s.Execute(
ctx, readTx, query, queryParams,
options.WithCollectStatsModeBasic(),
)
if err != nil {
return fmt.Errorf("executing: %w", err)
}
xcontext.LogDebug(ctx, r.log, "checking")
if !res.NextResultSet(ctx) || !res.HasNextRow() {
return nil
}
xcontext.LogDebug(ctx, r.log, "scanning")
for res.NextRow() {
var cdb courseDB
err = res.ScanNamed(cdb.getNamedValues()...)
if err != nil {
return fmt.Errorf("scanning row: %w", err)
}
courses = append(courses, mapCourseDB(cdb))
}
if err = res.Err(); err != nil {
return err
}
result.NextPageToken = courses[len(courses)-1].ID
xcontext.LogDebug(ctx, r.log, "scanned rows", slog.Int("count", len(courses)))
return nil
},
table.WithIdempotent())
if err != nil {
return nil, err
return domain.ListCoursesResult{}, err
}
return courses, err
result.Courses = courses
return result, err
}
func (r *ydbCourseRepository) Get(ctx context.Context, id string) (course domain.Course, err error) {
@ -579,9 +580,18 @@ type queryTemplateParams struct {
Suffix string
}
const queryTemplateSelect = `
{{ range .Declares }}DECLARE ${{.Name}} AS {{.Type}}\n{{end}}
func (p queryTemplateParams) render() (string, error) {
var sb strings.Builder
sb.Grow(len(queryTemplateSelect) * 3)
err := querySelect.Execute(&sb, p)
return sb.String(), err
}
const queryTemplateSelect = `{{ range .Declares }}DECLARE ${{.Name}} AS {{.Type}};{{end}}
SELECT {{.Fields}}
FROM {{.Table}}
WHERE {{ range .Conditions }}{{.}}\n{{end}}
WHERE {{ range .Conditions }}{{.}}{{end}}
{{.Suffix}}`
var querySelect = template.Must(template.New("").Parse(queryTemplateSelect))